Luís Murta
e7abae0d17
`add` method replaced with `insert`. `insert` and `select` implemented for new database base class. Database unit test added. Due to SQLite implementation of the primary key autoinc, the type of the IDs on the database for SQLite changed to Integer. https://www.sqlite.org/autoinc.html
71 lines
2.2 KiB
Python
71 lines
2.2 KiB
Python
from sqlalchemy import Engine, create_engine, select, text
|
|
from sqlalchemy.orm import Session, sessionmaker
|
|
from typing import Any, Optional, Sequence, Type, TypeVar
|
|
from pfbudget.db.exceptions import InsertError, SelectError
|
|
|
|
from pfbudget.db.model import Transaction
|
|
|
|
|
|
class Client:
|
|
def __init__(self, url: str, **kwargs: dict[str, Any]) -> None:
|
|
assert url, "Database URL is empty!"
|
|
self._engine = create_engine(url, **kwargs)
|
|
self._sessionmaker: Optional[sessionmaker[Session]] = None
|
|
|
|
def insert(
|
|
self, transactions: Sequence[Transaction], session: Optional[Session] = None
|
|
) -> None:
|
|
if not session:
|
|
with self.session as session_:
|
|
try:
|
|
session_.add_all(transactions)
|
|
except Exception as e:
|
|
session_.rollback()
|
|
raise InsertError(e)
|
|
else:
|
|
session_.commit()
|
|
else:
|
|
try:
|
|
session.add_all(transactions)
|
|
except Exception as e:
|
|
session.rollback()
|
|
raise InsertError(e)
|
|
else:
|
|
session.commit()
|
|
|
|
T = TypeVar("T")
|
|
|
|
def select(self, what: Type[T], session: Optional[Session] = None) -> Sequence[T]:
|
|
stmt = select(what)
|
|
result: Sequence[what] = []
|
|
|
|
if not session:
|
|
with self.session as session_:
|
|
try:
|
|
result = session_.scalars(stmt).all()
|
|
except Exception as e:
|
|
session_.rollback()
|
|
raise SelectError(e)
|
|
else:
|
|
session_.commit()
|
|
else:
|
|
try:
|
|
result = session.scalars(stmt).all()
|
|
except Exception as e:
|
|
session.rollback()
|
|
raise SelectError(e)
|
|
else:
|
|
session.commit()
|
|
|
|
return result
|
|
|
|
@property
|
|
def engine(self) -> Engine:
|
|
return self._engine
|
|
|
|
@property
|
|
def session(self) -> Session:
|
|
if not self._sessionmaker:
|
|
self._sessionmaker = sessionmaker(self._engine)
|
|
return self._sessionmaker()
|