QOL & test improvements

README.md

@@ -64,8 +64,6 @@ if __name__ == '__main__':
     uvicorn.run(app, host='0.0.0.0', port=8080)
 ```
 
- *See  [tests/integration/example/app.py](tests/integration/example/app.py)
-
 ## Example for pytest
 
 **conftest.py**

database_setup_tools/session_manager.py

@@ -1,22 +1,24 @@
-import sqlalchemy as sqla
 import threading
+from functools import cached_property
+from typing import Iterator, Optional
+
+import sqlalchemy as sqla
 from sqlalchemy.engine import Engine
 from sqlalchemy.orm import sessionmaker
 from sqlalchemy.orm.scoping import ScopedSession, scoped_session
-from typing import Iterator
 
 
 class SessionManager:
     """ Manages engines, sessions and connection pools. Thread-safe singleton """
-    _instance = None
+    _instances = []
     _lock = threading.Lock()
 
     def __new__(cls, *args, **kwargs):
-        if not cls._instance:
+        if not cls._get_cached_instance(args, kwargs):
             with cls._lock:
-                if not cls._instance:
-                    cls._instance = super(SessionManager, cls).__new__(cls)
-        return cls._instance
+                if not cls._get_cached_instance(args, kwargs):
+                    cls._instances.append((super(cls, cls).__new__(cls), (args, kwargs)))
+        return cls._get_cached_instance(args, kwargs)
 
     def __init__(self, database_uri: str, **kwargs):
         """ Session Manager constructor
@@ -32,17 +34,20 @@ def __init__(self, database_uri: str, **kwargs):
                 max_overflow (int): The maximum number of connections to the database
                 pre_ping (bool): Whether to ping the database before each connection, may fix connection issues
         """
+        if not isinstance(database_uri, str):
+            raise TypeError("database_uri must be a string")
+
         self._database_uri = database_uri
         self._engine = self._get_engine(**kwargs)
         self._session_factory = sessionmaker(self.engine)
         self._Session = scoped_session(self._session_factory)
 
-    @property
+    @cached_property
     def database_uri(self) -> str:
         """ Getter for the database URI """
         return self._database_uri
 
-    @property
+    @cached_property
     def engine(self) -> Engine:
         """ Getter for the engine """
         return self._engine
@@ -55,3 +60,11 @@ def get_session(self) -> Iterator[ScopedSession]:
     def _get_engine(self, **kwargs) -> Engine:
         """ Provides a database engine """
         return sqla.create_engine(self.database_uri, **kwargs)
+
+    @classmethod
+    def _get_cached_instance(cls, args: tuple, kwargs: dict) -> Optional[object]:
+        """ Provides a cached instance of the SessionManager class if existing """
+        for instance, arguments in cls._instances:
+            if arguments == (args, kwargs):
+                return instance
+        return None

database_setup_tools/setup.py

@@ -1,4 +1,5 @@
 import threading
+from typing import Optional
 
 import sqlalchemy_utils
 from sqlalchemy import MetaData
@@ -8,15 +9,15 @@
 
 class DatabaseSetup:
     """ Create the database and the tables if not done yet """
-    _instance = None
+    _instances = []
     _lock = threading.Lock()
 
     def __new__(cls, *args, **kwargs):
-        if not cls._instance:
+        if not cls._get_cached_instance(args, kwargs):
             with cls._lock:
-                if not cls._instance:
-                    cls._instance = super(DatabaseSetup, cls).__new__(cls)
-        return cls._instance
+                if not cls._get_cached_instance(args, kwargs):
+                    cls._instances.append((super(cls, cls).__new__(cls), (args, kwargs)))
+        return cls._get_cached_instance(args, kwargs)
 
     def __init__(self, model_metadata: MetaData, database_uri: str):
         """ Set up a database based on its URI and metadata. Will not overwrite existing data.
@@ -26,6 +27,12 @@ def __init__(self, model_metadata: MetaData, database_uri: str):
             database_uri (str): The URI of the database to create the tables for
 
         """
+        if not isinstance(model_metadata, MetaData):
+            raise TypeError("model_metadata must be a MetaData")
+
+        if not isinstance(database_uri, str):
+            raise TypeError("database_uri must be a string")
+
         self._model_metadata = model_metadata
         self._database_uri = database_uri
         self.create_database()
@@ -64,3 +71,11 @@ def create_database(self):
         sqlalchemy_utils.create_database(self.database_uri)
         session_manager = SessionManager(self.database_uri)
         self.model_metadata.create_all(session_manager.engine)
+
+    @classmethod
+    def _get_cached_instance(cls, args: tuple, kwargs: dict) -> Optional[object]:
+        """ Provides a cached instance of the SessionManager class if existing """
+        for instance, arguments in cls._instances:
+            if arguments == (args, kwargs):
+                return instance
+        return None