fix: 🐛 Raw db session requirement if tenant id is passed

.github/copilot-instructions.md

@@ -43,15 +43,15 @@ Always reference these instructions first and fallback to search or bash command
    python -c "
    from sql_db_utils import SQLSessionManager
    manager = SQLSessionManager()
-   
+
    @manager.register_precreate('test_db')
    def test_precreate(tenant_id):
        return 'SELECT 1;'
-   
+
    @manager.register_postcreate('test_db')
    def test_postcreate(tenant_id):
        return 'SELECT 2;'
-   
+
    print('Precreate/Postcreate registration successful')
    "
    ```
@@ -117,7 +117,7 @@ sql-db-utils/
 ### Development Dependencies:
 - **Testing**: pytest, pytest-cov, coverage
 - **Linting**: ruff (replaces black, flake8, isort)
-- **Git hooks**: pre-commit 
+- **Git hooks**: pre-commit
 - **Type checking**: Built into package development
 - **Core Dependencies**: SQLAlchemy, sqlalchemy-utils, psycopg, python-dateutil, whenever
 
@@ -221,4 +221,4 @@ sql-db-utils/
 - Changes to sync version should be mirrored in async version
 - Test both implementations when making session management changes
 - Verify async patterns use proper `await` keywords
-- Check that both versions handle errors consistently
+- Check that both versions handle errors consistently

.pre-commit-config.yaml

@@ -1,12 +1,12 @@
 repos:
 -   repo: https://github.com/pre-commit/pre-commit-hooks
-    rev: v5.0.0
+    rev: v6.0.0
     hooks:
     -   id: end-of-file-fixer
     -   id: trailing-whitespace
     -   id: requirements-txt-fixer
 -   repo: https://github.com/charliermarsh/ruff-pre-commit
-    rev: v0.11.0
+    rev: v0.14.4
     hooks:
     -   id: ruff
         args:

sql_db_utils/__version__.py

@@ -1 +1 @@
-__version__ = "1.3.1"
+__version__ = "1.3.2"

sql_db_utils/asyncio/session_management.py

@@ -121,11 +121,27 @@ async def get_engine_obj(
     ) -> AsyncEngine:
         return await self._get_engine(database=database, tenant_id=tenant_id, metadata=metadata)
 
-    def get_db_factory(self, database: str, retrying: bool = False) -> AsyncGenerator[AsyncSession, Any]:
+    def get_db_factory(
+        self, database: str, retrying: bool = False, raw_db: bool = False
+    ) -> AsyncGenerator[AsyncSession, Any]:
+        """
+        Create a database session factory for FastAPI dependency injection.
+
+        Args:
+            database: Name of the database.
+            retrying: Whether to enable retry logic for queries.
+            raw_db: If True, bypasses tenant_id and creates a system-level session without tenant isolation.
+                Use with caution for admin operations, migrations, or other system-level tasks.
+                WARNING: Bypassing tenant isolation may expose all data in the database to the session.
+                Only use raw_db=True for trusted, internal operations.
+
+        Returns:
+            An async factory function that yields database sessions.
+        """
         from fastapi import Cookie
 
         async def get_db(tenant_id: Annotated[Union[str, None], Cookie()] = None) -> AsyncGenerator[AsyncSession, Any]:
-            yield await self.get_session(database=database, tenant_id=tenant_id, retrying=retrying)
+            yield await self.get_session(database=database, tenant_id=None if raw_db else tenant_id, retrying=retrying)
 
         return get_db
 

sql_db_utils/session_management.py

@@ -121,11 +121,24 @@ def get_engine_obj(
     ) -> Engine:
         return self._get_engine(database=database, tenant_id=tenant_id, metadata=metadata)
 
-    def get_db_factory(self, database: str, retrying: bool = False) -> Callable:
+    def get_db_factory(self, database: str, retrying: bool = False, raw_db: bool = False) -> Callable:
+        """
+        Create a database session factory for FastAPI dependency injection.
+
+        Args:
+            database (str): Name of the database.
+            retrying (bool): Whether to enable retry logic for queries.
+            raw_db (bool): If True, bypasses tenant_id and creates a system-level session without tenant isolation.
+                Use with caution for admin operations, migrations, or other system-level tasks.
+                WARNING: Bypassing tenant isolation may expose all data and should only be used by trusted code.
+
+        Returns:
+            Callable: A factory function that yields database sessions.
+        """
         from fastapi import Cookie
 
         def get_db(tenant_id: Annotated[str, Cookie()] = None) -> Session:
-            yield self.get_session(database=database, tenant_id=tenant_id, retrying=retrying)
+            yield self.get_session(database=database, tenant_id=None if raw_db else tenant_id, retrying=retrying)
 
         return get_db