add user table
All checks were successful
Backend CI / unit-test (push) Successful in 48s

This commit is contained in:
2025-09-14 15:40:11 +02:00
parent 0a906535dc
commit 479d5cd230
7 changed files with 253 additions and 78 deletions

View File

@@ -15,22 +15,43 @@ if TYPE_CHECKING:
class Database:
def __init__(self, database_url: str | None = None, *, echo: bool = False, connect_args: dict | None = None) -> None:
def __init__(
self,
database_url: str | None = None,
*,
echo: bool = False,
connect_args: dict | None = None,
) -> None:
self._database_url = database_url or "sqlite:///:memory:"
default_connect = {"check_same_thread": False, "timeout": 30} if self._database_url.startswith("sqlite") else {}
default_connect = (
{"check_same_thread": False, "timeout": 30}
if self._database_url.startswith("sqlite")
else {}
)
merged_connect = {**default_connect, **(connect_args or {})}
if self._database_url == "sqlite:///:memory:":
logger = logging.getLogger(__name__)
logger.warning("Using in-memory SQLite database; all data will be lost when the application stops.")
self._engine = create_engine(self._database_url, echo=echo, connect_args=merged_connect, poolclass=StaticPool)
logger.warning(
"Using in-memory SQLite database; all data will be lost when the application stops."
)
self._engine = create_engine(
self._database_url,
echo=echo,
connect_args=merged_connect,
poolclass=StaticPool,
)
else:
self._engine = create_engine(self._database_url, echo=echo, connect_args=merged_connect)
self._engine = create_engine(
self._database_url, echo=echo, connect_args=merged_connect
)
if self._database_url.startswith("sqlite"):
def _enable_sqlite_pragmas(dbapi_conn: DBAPIConnection, _connection_record: object) -> None:
def _enable_sqlite_pragmas(
dbapi_conn: DBAPIConnection, _connection_record: object
) -> None:
try:
cur = dbapi_conn.cursor()
cur.execute("PRAGMA journal_mode=WAL;")
@@ -62,5 +83,10 @@ class Database:
self._engine.dispose()
def create_database(database_url: str | None = None, *, echo: bool = False, connect_args: dict | None = None) -> Database:
def create_database(
database_url: str | None = None,
*,
echo: bool = False,
connect_args: dict | None = None,
) -> Database:
return Database(database_url, echo=echo, connect_args=connect_args)