Skip to content

fractal_server.app.db

db module, loosely adapted from https://testdriven.io/blog/fastapi-sqlmodel/#async-sqlmodel

CLASS DESCRIPTION
DB

DB class

Attributes

Classes

DB

DB class

METHOD DESCRIPTION
get_async_db

Get async database session
get_sync_db

Get sync database session
Source code in fractal_server/app/db/__init__.py 
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
class DB:
    """
    DB class
    """

    @classmethod
    def engine_async(cls) -> AsyncEngine:
        try:
            return cls._engine_async
        except AttributeError:
            cls.set_async_db()
            return cls._engine_async

    @classmethod
    def engine_sync(cls) -> Engine:
        try:
            return cls._engine_sync
        except AttributeError:
            cls.set_sync_db()
            return cls._engine_sync

    @classmethod
    def set_async_db(cls) -> None:
        db_settings = Inject(get_db_settings)

        cls._engine_async = create_async_engine(
            db_settings.DATABASE_URL,
            echo=(db_settings.DB_ECHO == "true"),
            future=True,
            pool_pre_ping=True,
        )
        cls._async_session_maker = async_sessionmaker(
            cls._engine_async,
            class_=AsyncSession,
            expire_on_commit=False,
        )

    @classmethod
    def set_sync_db(cls) -> None:
        db_settings = Inject(get_db_settings)

        cls._engine_sync = create_engine(
            db_settings.DATABASE_URL,
            echo=(db_settings.DB_ECHO == "true"),
            future=True,
            pool_pre_ping=True,
        )

        cls._sync_session_maker = sessionmaker(cls._engine_sync)

    @classmethod
    async def get_async_db(cls) -> AsyncGenerator[AsyncSession, None]:
        """
        Get async database session
        """
        try:
            session_maker = cls._async_session_maker()
        except AttributeError:
            cls.set_async_db()
            session_maker = cls._async_session_maker()
        async with session_maker as async_session:
            yield async_session

    @classmethod
    def get_sync_db(cls) -> Generator[DBSyncSession, None, None]:
        """
        Get sync database session
        """
        try:
            session_maker = cls._sync_session_maker()
        except AttributeError:
            cls.set_sync_db()
            session_maker = cls._sync_session_maker()
        with session_maker as sync_session:
            yield sync_session

Methods:

get_async_db() async classmethod

Get async database session

Source code in fractal_server/app/db/__init__.py 
75
76
77
78
79
80
81
82
83
84
85
86
@classmethod
async def get_async_db(cls) -> AsyncGenerator[AsyncSession, None]:
    """
    Get async database session
    """
    try:
        session_maker = cls._async_session_maker()
    except AttributeError:
        cls.set_async_db()
        session_maker = cls._async_session_maker()
    async with session_maker as async_session:
        yield async_session
get_sync_db() classmethod

Get sync database session

Source code in fractal_server/app/db/__init__.py 
88
89
90
91
92
93
94
95
96
97
98
99
@classmethod
def get_sync_db(cls) -> Generator[DBSyncSession, None, None]:
    """
    Get sync database session
    """
    try:
        session_maker = cls._sync_session_maker()
    except AttributeError:
        cls.set_sync_db()
        session_maker = cls._sync_session_maker()
    with session_maker as sync_session:
        yield sync_session

Functions: