Skip to content

Instantly share code, notes, and snippets.

@turkishmaid
Last active July 15, 2022 11:21
Show Gist options
  • Save turkishmaid/b8d19df0122566a8b164e370bad10aab to your computer and use it in GitHub Desktop.
Save turkishmaid/b8d19df0122566a8b164e370bad10aab to your computer and use it in GitHub Desktop.
How to auto-close a MongoDB connection
#!/usr/bin/env python
# coding: utf-8
"""
Handle MongoDB connection.
"""
# Created: 01.11.21
from time import perf_counter, sleep
import logging
import atexit
from typing import Union
from pymongo import MongoClient
from pymongo.database import Database, Collection
from base import the_config
from pythonx import split_seconds
logger = logging.getLogger(__name__)
mongo_client: Union[MongoClient, None] = None
def bye():
global mongo_client, mongo_db
if mongo_client:
mongo_client.close()
mongo_client = None
mongo_db = None
dt = perf_counter() - mongo_pc0
logger.info(f"closed mongo connection after {split_seconds(dt)}")
# sleep(2.0) # -> ja, funktionert auch wenn man den Debugger neustartet :)
atexit.register(bye)
logger.info("atexit.register() for MongoDB connection.")
mongo_uri = the_config["mongo"]["uri"]
mongo_dbname = the_config["mongo"]["db"]
logger.info(f"connecting to {mongo_uri}")
mongo_client = MongoClient(mongo_uri)
logger.info(f"choosing database {mongo_dbname}")
mongo_db: Database = mongo_client[mongo_dbname]
mongo_pc0 = perf_counter()
def get_mongo_db(coll: str = None) -> Union[Database, Collection]:
if coll:
assert isinstance(coll, str)
return mongo_db[coll]
else:
return mongo_db
def coll(symbol: str) -> Collection:
return mongo_db[the_config["coll"][symbol]]
#
# pythonx/split_secnds()
#
def split_seconds(seconds: float) -> str:
"""
Funny stuff: render a duration like 5d 10m 3.5s.
:param seconds:
:return:
"""
secs = seconds % 60
seconds //= 60
seconds = int(round(seconds, 0))
mins = seconds % 60
seconds //= 60
hrs = seconds % 24
seconds //= 24
days = seconds
a = []
if days:
a.append(f"{days}d")
if hrs:
a.append(f"{hrs}h")
if mins:
a.append(f"{mins}m")
if secs >= 0.1:
a.append(f"{secs:0.1f}s")
if a:
return " ".join(a)
else:
return "no time"
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment