v1.0.17

README.md

@@ -50,8 +50,11 @@ pip install ddcDatabases[pgsql]
 # SQLITE
 ```
 class Sqlite(
-    file_path: Optional[str] = None,
+    filepath: Optional[str] = None,
     echo: Optional[bool] = None,
+    autoflush: Optional[bool] = None,
+    expire_on_commit: Optional[bool] = None,
+    extra_engine_args: Optional[dict] = None,
 )
 ```
 
@@ -83,13 +86,16 @@ with Sqlite().engine() as engine:
 class MSSQL(        
     host: Optional[str] = None,
     port: Optional[int] = None,
-    username: Optional[str] = None,
+    user: Optional[str] = None,
     password: Optional[str] = None,
     database: Optional[str] = None,
     schema: Optional[str] = None,
     echo: Optional[bool] = None,
     pool_size: Optional[int] = None,
-    max_overflow: Optional[int] = None
+    max_overflow: Optional[int] = None,
+    autoflush: Optional[bool] = None,
+    expire_on_commit: Optional[bool] = None,
+    extra_engine_args: Optional[dict] = None,
 )
 ```
 
@@ -141,10 +147,13 @@ async with MSSQL().async_engine() as engine:
 class DBPostgres(
     host: Optional[str] = None,
     port: Optional[int] = None,
-    username: Optional[str] = None,
+    user: Optional[str] = None,
     password: Optional[str] = None,
     database: Optional[str] = None,
     echo: Optional[bool] = None,
+    autoflush: Optional[bool] = None,
+    expire_on_commit: Optional[bool] = None,
+    engine_args: Optional[dict] = None,
 )
 ```
 

ddcDatabases/db_utils.py

@@ -23,7 +23,6 @@
 
 
 class BaseConn:
-
     def __init__(
         self,
         connection_url,
@@ -80,7 +79,10 @@ async def __aexit__(self, exc_type, exc_val, exc_tb):
 
     @contextmanager
     def engine(self) -> Generator:
-        _connection_url = URL.create(drivername=self.sync_driver, **self.connection_url)
+        _connection_url = URL.create(
+            drivername=self.sync_driver,
+            **self.connection_url,
+        )
         _engine_args = {
             "url": _connection_url,
             **self.engine_args,
@@ -91,7 +93,10 @@ def engine(self) -> Generator:
 
     @asynccontextmanager
     async def async_engine(self) -> AsyncGenerator:
-        _connection_url = URL.create(drivername=self.async_driver, **self.connection_url)
+        _connection_url = URL.create(
+            drivername=self.async_driver,
+            **self.connection_url,
+        )
         _engine_args = {
             "url": _connection_url,
             **self.engine_args,
@@ -102,51 +107,77 @@ async def async_engine(self) -> AsyncGenerator:
 
     def _test_connection_sync(self, session: Session) -> None:
         del self.connection_url["password"]
-        _connection_url = URL.create(**self.connection_url, drivername=self.sync_driver)
-        test_connection = TestConnections(sync_session=session, host_url=_connection_url)
+        _connection_url = URL.create(
+            **self.connection_url,
+            drivername=self.sync_driver,
+        )
+        test_connection = TestConnections(
+            sync_session=session,
+            host_url=_connection_url,
+        )
         test_connection.test_connection_sync()
 
     async def _test_connection_async(self, session: AsyncSession) -> None:
         del self.connection_url["password"]
-        _connection_url = URL.create(**self.connection_url, drivername=self.async_driver)
-        test_connection = TestConnections(async_session=session, host_url=_connection_url)
+        _connection_url = URL.create(
+            **self.connection_url,
+            drivername=self.async_driver,
+        )
+        test_connection = TestConnections(
+            async_session=session,
+            host_url=_connection_url,
+        )
         await test_connection.test_connection_async()
 
 
 class TestConnections:
-
     def __init__(
         self,
         sync_session: Session = None,
         async_session: AsyncSession = None,
         host_url: URL = "",
     ):
-        self.dt = datetime.now().strftime("%Y-%m-%dT%H:%M:%S.%f")[:-3]
         self.sync_session = sync_session
         self.async_session = async_session
         self.host_url = host_url
+        self.dt = datetime.now().strftime("%Y-%m-%dT%H:%M:%S.%f")[:-3]
+        self.successful_msg = "[INFO]:Connection to database successful"
+        self.failed_msg = "[ERROR]:Connection to database failed"
 
     def test_connection_sync(self) -> None:
         try:
             self.sync_session.execute(sa.text("SELECT 1"))
-            sys.stdout.write(f"[{self.dt}]:[INFO]:Connection to database successful | {self.host_url}\n")
+            sys.stdout.write(
+                f"[{self.dt}]:{self.successful_msg} | "
+                f"{self.host_url}\n"
+            )
         except Exception as e:
             self.sync_session.close()
-            sys.stderr.write(f"[{self.dt}]:[ERROR]:Connection to datatabse failed | {self.host_url} | {repr(e)}\n")
+            sys.stderr.write(
+                f"[{self.dt}]:{self.failed_msg} | "
+                f"{self.host_url} | "
+                f"{repr(e)}\n"
+            )
             raise
 
     async def test_connection_async(self) -> None:
         try:
             await self.async_session.execute(sa.text("SELECT 1"))
-            sys.stdout.write(f"[{self.dt}]:[INFO]:Connection to database successful | {self.host_url}\n")
+            sys.stdout.write(
+                f"[{self.dt}]:{self.successful_msg} | "
+                f"{self.host_url}\n"
+            )
         except Exception as e:
             await self.async_session.close()
-            sys.stderr.write(f"[{self.dt}]:[ERROR]:Connection to datatabse failed | {self.host_url} | {repr(e)}\n")
+            sys.stderr.write(
+                f"[{self.dt}]:{self.failed_msg} | "
+                f"{self.host_url} | "
+                f"{repr(e)}\n"
+            )
             raise
 
 
 class DBUtils:
-
     def __init__(self, session):
         self.session = session
 
@@ -211,7 +242,6 @@ def execute(self, stmt) -> None:
 
 
 class DBUtilsAsync:
-
     def __init__(self, session):
         self.session = session
 

ddcDatabases/exceptions.py

@@ -1,10 +1,12 @@
 # -*- encoding: utf-8 -*-
 import sys
+from datetime import datetime
 
 
 class CustomBaseException(Exception):
     def __init__(self, msg):
-        sys.stderr.write(repr(msg))
+        dt = datetime.now().strftime("%Y-%m-%dT%H:%M:%S.%f")[:-3]
+        sys.stderr.write(f"[{dt}]:[ERROR]:{repr(msg)}")
 
 
 class DBFetchAllException(CustomBaseException):

ddcDatabases/mssql.py

@@ -25,6 +25,7 @@ def __init__(
         max_overflow: Optional[int] = None,
         autoflush: Optional[bool] = None,
         expire_on_commit: Optional[bool] = None,
+        extra_engine_args: Optional[dict] = None,
     ):
         _settings = MSSQLSettings()
         if not _settings.user or not _settings.password:
@@ -41,7 +42,7 @@ def __init__(
         self.odbcdriver_version = int(_settings.odbcdriver_version)
         self.connection_url = {
             "host": host or _settings.host,
-            "port":  int(port or _settings.port),
+            "port": int(port or _settings.port),
             "database": database or _settings.database,
             "username": user or _settings.user,
             "password": password or _settings.password,
@@ -50,10 +51,12 @@ def __init__(
                 "TrustServerCertificate": "yes",
             },
         }
+        self.extra_engine_args = extra_engine_args or {}
         self.engine_args = {
             "pool_size": self.pool_size,
             "max_overflow": self.max_overflow,
             "echo": self.echo,
+            **self.extra_engine_args,
         }
 
         super().__init__(
@@ -73,7 +76,10 @@ def _test_connection_sync(self, session: Session) -> None:
             drivername=self.sync_driver,
             query={"schema": self.schema},
         )
-        test_connection = TestConnections(sync_session=session, host_url=_connection_url)
+        test_connection = TestConnections(
+            sync_session=session,
+            host_url=_connection_url,
+        )
         test_connection.test_connection_sync()
 
     async def _test_connection_async(self, session: AsyncSession) -> None:
@@ -84,5 +90,8 @@ async def _test_connection_async(self, session: AsyncSession) -> None:
             drivername=self.async_driver,
             query={"schema": self.schema},
         )
-        test_connection = TestConnections(async_session=session, host_url=_connection_url)
+        test_connection = TestConnections(
+            async_session=session,
+            host_url=_connection_url,
+        )
         await test_connection.test_connection_async()

ddcDatabases/postgresql.py

@@ -19,6 +19,7 @@ def __init__(
         echo: Optional[bool] = None,
         autoflush: Optional[bool] = None,
         expire_on_commit: Optional[bool] = None,
+        extra_engine_args: Optional[dict] = None,
     ):
         _settings = PostgreSQLSettings()
         if not _settings.user or not _settings.password:
@@ -31,13 +32,15 @@ def __init__(
         self.sync_driver = _settings.sync_driver
         self.connection_url = {
             "host": host or _settings.host,
-            "port":  int(port or _settings.port),
+            "port": int(port or _settings.port),
             "database": database or _settings.database,
             "username": user or _settings.user,
             "password": password or _settings.password,
         }
+        self.extra_engine_args = extra_engine_args or {}
         self.engine_args = {
             "echo": self.echo,
+            **self.extra_engine_args,
         }
 
         super().__init__(

ddcDatabases/sqlite.py

@@ -15,21 +15,29 @@ class Sqlite:
 
     def __init__(
         self,
-        file_path: Optional[str] = None,
+        filepath: Optional[str] = None,
         echo: Optional[bool] = None,
+        autoflush: Optional[bool] = None,
+        expire_on_commit: Optional[bool] = None,
+        extra_engine_args: Optional[dict] = None,
     ):
         _settings = SQLiteSettings()
+        self.filepath = filepath or _settings.file_path
+        self.echo = echo or _settings.echo
+        self.autoflush = autoflush
+        self.expire_on_commit = expire_on_commit
+        self.extra_engine_args = extra_engine_args or {}
         self.temp_engine = None
         self.session = None
-        self.file_path = file_path or _settings.file_path
-        self.echo = echo or _settings.echo
 
     def __enter__(self):
         with self.engine() as self.temp_engine:
-            session_maker = sessionmaker(bind=self.temp_engine,
-                                         class_=Session,
-                                         autoflush=True,
-                                         expire_on_commit=True)
+            session_maker = sessionmaker(
+                bind=self.temp_engine,
+                class_=Session,
+                autoflush=self.autoflush or True,
+                expire_on_commit=self.expire_on_commit or True,
+            )
 
         with session_maker.begin() as self.session:
             return self.session
@@ -44,17 +52,17 @@ def __exit__(self, exc_type, exc_val, exc_tb):
     def engine(self) -> Engine | None:
         try:
             _engine_args = {
-                "url": f"sqlite:///{self.file_path}",
+                "url": f"sqlite:///{self.filepath}",
                 "echo": self.echo,
+                **self.extra_engine_args,
             }
             _engine = create_engine(**_engine_args)
             yield _engine
             _engine.dispose()
         except Exception as e:
             dt = datetime.now().strftime("%Y-%m-%dT%H:%M:%S.%f")[:-3]
             sys.stderr.write(
-                f"[{dt}]:"
-                "[ERROR]:Unable to Create Database Engine | "
-                f"{repr(e)}"
+                f"[{dt}]:[ERROR]:Unable to Create Database Engine | "
+                f"{repr(e)}\n"
             )
             raise

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "poetry.core.masonry.api"
 
 [tool.poetry]
 name = "ddcDatabases"
-version = "1.0.16"
+version = "1.0.17"
 description = "Databases Connection and Queries"
 license = "MIT"
 readme = "README.md"

tests/conftest.py

@@ -1,12 +1,17 @@
 # -*- coding: utf-8 -*-
 import pytest
+from sqlalchemy.pool import StaticPool
 from ddcDatabases import Sqlite
-from tests.data.base_data import sqlite_filename, get_fake_test_data
+from tests.data.base_data import get_fake_test_data, sqlite_filename
 
 
 @pytest.fixture(name="sqlite_session", scope="session")
 def sqlite_session():
-    with Sqlite(sqlite_filename) as session:
+    extra_engine_args = {"poolclass": StaticPool}
+    with Sqlite(
+        filepath=sqlite_filename,
+        extra_engine_args=extra_engine_args,
+    ) as session:
         yield session
 
 

tests/dal/test_model_dal.py

@@ -5,7 +5,7 @@
 from tests.models.test_model import ModelTest
 
 
-class TestModelDal:
+class ModelDalTest:
     """ Data Abstraction Layer """
 
     def __init__(self, db_session):