Skip to content

Database operators

commons.database_operators

Database connection and settings operators.

Functions:

Name Description
create_engine

create_engine 

create_engine() -> Engine
Source code in src/commons/database_operators.py 
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
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
def create_engine() -> Engine:
    database_settings = settings.DatabaseSettings.read()

    if database_settings.mode == "mssql":
        assert database_settings.mssql is not None, "MSSQL database settings must be present, if mode is mssql."

        default_query = {
            "driver": "ODBC Driver 18 for SQL Server",
            "Encrypt": "yes",
            "TrustServerCertificate": "no",
            "Connection Timeout": "30",
        }
        query = {**default_query, **(database_settings.mssql.query or {})}

        url = URL.create(
            "mssql+pyodbc",
            username=database_settings.mssql.username,
            password=database_settings.mssql.password,
            host=database_settings.mssql.server,
            port=database_settings.mssql.port,
            database=database_settings.mssql.database,
            query=query,
        )

        return sql_alchemy_create_engine(url, future=True)

    else:
        app_settings = settings.AppSettings.read()

        if database_settings.local and database_settings.local.path:
            path = Path(database_settings.local.path)
            if path.is_absolute():
                database_file = path
            else:
                database_file = (app_settings.data_dir / path).absolute()
        else:
            database_file = (app_settings.data_dir / "database.db").absolute()

        database_file.parent.mkdir(parents=True, exist_ok=True)

        return sql_alchemy_create_engine(f"sqlite:///{str(database_file)}")