Last active
July 15, 2022 11:21
-
-
Save turkishmaid/b8d19df0122566a8b164e370bad10aab to your computer and use it in GitHub Desktop.
How to auto-close a MongoDB connection
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
#!/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