Skip to content

Utils

create_secmaster_db(db_path, schema_version=1)

Create a new security master SQLite database using a selected schema version.

The database file is created at the given path and initialized by executing the SQL script located in the schema_versions directory adjacent to this module.

The function expects the schema script to set PRAGMA user_version to the corresponding schema version and verifies this after execution.

Parameters:

Name Type Description Default
db_path Path

Filesystem path at which the SQLite database file will be created.

 required
schema_version int

Version number selecting the schema script to apply.

 1

Returns:

Type Description
Path

The path to the created database file.

Raises:

Type Description
FileExistsError

If a file already exists at db_path.
FileNotFoundError

If the schema script for schema_version does not exist.
DatabaseError

If the applied schema does not set the expected user_version or if SQLite fails while executing the schema.
Source code in src/onesecondtrader/secmaster/utils.py 
 7
 8
 9
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
51
52
53
54
55
56
57
58
59
60
61
62
63
64
def create_secmaster_db(db_path: pathlib.Path, schema_version: int = 1) -> pathlib.Path:
    """
    Create a new security master SQLite database using a selected schema version.

    The database file is created at the given path and initialized by executing the SQL script located in the `schema_versions` directory adjacent to this module.

    The function expects the schema script to set `PRAGMA user_version` to the corresponding schema version and verifies this after execution.

    Parameters:
        db_path:
            Filesystem path at which the SQLite database file will be created.
        schema_version:
            Version number selecting the schema script to apply.

    Returns:
        The path to the created database file.

    Raises:
        FileExistsError:
            If a file already exists at `db_path`.
        FileNotFoundError:
            If the schema script for `schema_version` does not exist.
        sqlite3.DatabaseError:
            If the applied schema does not set the expected `user_version`
            or if SQLite fails while executing the schema.
    """
    if db_path.exists():
        raise FileExistsError(f"Database already exists: {db_path}")

    schema_path = (
        pathlib.Path(__file__).resolve().parent
        / "schema_versions"
        / f"secmaster_schema_v{schema_version}.sql"
    )

    if not schema_path.is_file():
        raise FileNotFoundError(
            f"Schema version {schema_version} not found: {schema_path}"
        )

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

    schema_sql = schema_path.read_text(encoding="utf-8")

    with sqlite3.connect(str(db_path)) as con:
        con.execute("PRAGMA foreign_keys = ON;")
        con.executescript(schema_sql)

        row = con.execute("PRAGMA user_version;").fetchone()
        actual_version = int(row[0]) if row else 0

        if actual_version != schema_version:
            raise sqlite3.DatabaseError(
                f"Schema script set user_version={actual_version}, "
                f"expected {schema_version}"
            )

    return db_path