fmt

services/db.py

@@ -5,8 +5,7 @@ import traceback
 import warnings
 from typing import Any, Callable, Dict, TypeVar
 
-from sqlalchemy import (JSON, Column, Engine, Integer, create_engine, event,
-                        exc, inspect)
+from sqlalchemy import JSON, Column, Engine, Integer, create_engine, event, exc, inspect
 from sqlalchemy.ext.declarative import declarative_base
 from sqlalchemy.orm import Session, configure_mappers
 from sqlalchemy.sql.schema import Table
@@ -19,13 +18,13 @@ from settings import DB_URL
 engine = create_engine(DB_URL, echo=False, pool_size=10, max_overflow=20)
 inspector = inspect(engine)
 configure_mappers()
-T = TypeVar('T')
+T = TypeVar("T")
 REGISTRY: Dict[str, type] = {}
-FILTERED_FIELDS = ['_sa_instance_state', 'search_vector']
+FILTERED_FIELDS = ["_sa_instance_state", "search_vector"]
 
 
 # noinspection PyUnusedLocal
-def local_session(src=''):
+def local_session(src=""):
     return Session(bind=engine, expire_on_commit=False)
 
 
@@ -36,7 +35,7 @@ class Base(declarative_base()):
     __init__: Callable
     __allow_unmapped__ = True
     __abstract__ = True
-    __table_args__ = {'extend_existing': True}
+    __table_args__ = {"extend_existing": True}
 
     id = Column(Integer, primary_key=True)
 
@@ -57,11 +56,11 @@ class Base(declarative_base()):
                 else:
                     data[c] = value
             # Add synthetic field .stat
-            if hasattr(self, 'stat'):
-                data['stat'] = self.stat
+            if hasattr(self, "stat"):
+                data["stat"] = self.stat
             return data
         except Exception as e:
-            logger.error(f'Error occurred while converting object to dictionary: {e}')
+            logger.error(f"Error occurred while converting object to dictionary: {e}")
             return {}
 
     def update(self, values: Dict[str, Any]) -> None:
@@ -79,22 +78,22 @@ def warning_with_traceback(
     message: Warning | str, category, filename: str, lineno: int, file=None, line=None
 ):
     tb = traceback.format_stack()
-    tb_str = ''.join(tb)
-    return f'{message} ({filename}, {lineno}): {category.__name__}\n{tb_str}'
+    tb_str = "".join(tb)
+    return f"{message} ({filename}, {lineno}): {category.__name__}\n{tb_str}"
 
 
 # Установка функции вывода трейсбека для предупреждений SQLAlchemy
 warnings.showwarning = warning_with_traceback
-warnings.simplefilter('always', exc.SAWarning)
+warnings.simplefilter("always", exc.SAWarning)
 
 
-@event.listens_for(Engine, 'before_cursor_execute')
+@event.listens_for(Engine, "before_cursor_execute")
 def before_cursor_execute(conn, cursor, statement, parameters, context, executemany):
     conn.query_start_time = time.time()
-    conn.last_statement = ''
+    conn.last_statement = ""
 
 
-@event.listens_for(Engine, 'after_cursor_execute')
+@event.listens_for(Engine, "after_cursor_execute")
 def after_cursor_execute(conn, cursor, statement, parameters, context, executemany):
     compiled_statement = context.compiled.string
     compiled_parameters = context.compiled.params
@@ -105,7 +104,6 @@ def after_cursor_execute(conn, cursor, statement, parameters, context, executema
         else:
             query = compiled_statement  # or handle this case in a way that makes sense for your application
 
-
         if elapsed > 1 and conn.last_statement != query:
             conn.last_statement = query
             logger.debug(f"\n{query}\n{'*' * math.floor(elapsed)} {elapsed:.3f} s\n")