/
artyuhovs
/
AIFeedBackTrainingBot
Обзор
Документация
Войти
/
artyuhovs
/
AIFeedBackTrainingBot
Код
Запросы
0
Задачи
Вики
Пакеты
0
Релизы
0
CI/CD
Аналитика
Безопасность
master
app/db/database.py
48 строк
1 KB
Codex
Завершить архитектурное оздоровление и исправить итоги
15 июл 2026, 11:05
15 июл 2026, 11:05
c65f9a6
Код
Авторство
О чём код?
from __future__ import annotations import asyncio import sqlite3 from collections.abc import Callable, Generator from contextlib import contextmanager from pathlib import Path from typing import TypeVar from app.db.session import connect from app.utils.time_utils import Clock, SystemClock ResultT = TypeVar("ResultT") class Database: """Own SQLite connection and transaction boundaries without imposing an ORM.""" def __init__(self, path: Path, clock: Clock | None = None) -> None: self.path = path self.clock = clock or SystemClock() self.path.parent.mkdir(parents=True, exist_ok=True) def connect(self) -> sqlite3.Connection: return connect(self.path) @contextmanager def transaction(self, *, immediate: bool = False) -> Generator[sqlite3.Connection, None, None]: connection = self.connect() try: connection.execute("BEGIN IMMEDIATE" if immediate else "BEGIN") yield connection connection.commit() except BaseException: connection.rollback() raise finally: connection.close() async def run(self, operation: Callable[[sqlite3.Connection], ResultT]) -> ResultT: def execute() -> ResultT: with self.transaction() as connection: return operation(connection) return await asyncio.to_thread(execute) def now_iso(self) -> str: return self.clock.now().isoformat()