coddard
60 lines
1.3 KiB
Python
60 lines
1.3 KiB
Python
import sqlite3
|
|
import threading
|
|
from typing import Optional
|
|
from ib_async import IB
|
|
|
|
_ib_instance: Optional[IB] = None
|
|
_db_conn: Optional[sqlite3.Connection] = None
|
|
_db_cursor: Optional[sqlite3.Cursor] = None
|
|
_db_write_lock: Optional[threading.Lock] = None
|
|
|
|
|
|
def get_ib() -> IB:
|
|
if _ib_instance is None:
|
|
raise RuntimeError("IB instance not initialized")
|
|
return _ib_instance
|
|
|
|
|
|
def set_ib_instance(ib: IB):
|
|
global _ib_instance
|
|
_ib_instance = ib
|
|
|
|
|
|
def get_db_conn() -> sqlite3.Connection:
|
|
if _db_conn is None:
|
|
raise RuntimeError("Database connection not initialized")
|
|
return _db_conn
|
|
|
|
|
|
def get_db_cursor() -> sqlite3.Cursor:
|
|
if _db_cursor is None:
|
|
raise RuntimeError("Database cursor not initialized")
|
|
return _db_cursor
|
|
|
|
|
|
def get_db_lock() -> threading.Lock:
|
|
if _db_write_lock is None:
|
|
raise RuntimeError("Database write lock not initialized")
|
|
return _db_write_lock
|
|
|
|
|
|
def set_database_dependencies(
|
|
conn: sqlite3.Connection,
|
|
cursor: sqlite3.Cursor,
|
|
lock: threading.Lock,
|
|
):
|
|
global _db_conn, _db_cursor, _db_write_lock
|
|
_db_conn = conn
|
|
_db_cursor = cursor
|
|
_db_write_lock = lock
|
|
|
|
|
|
def setup_dependencies(
|
|
ib: IB,
|
|
conn: sqlite3.Connection,
|
|
cursor: sqlite3.Cursor,
|
|
lock: threading.Lock,
|
|
):
|
|
set_ib_instance(ib)
|
|
set_database_dependencies(conn, cursor, lock)
|