118 lines
3.1 KiB
Python
118 lines
3.1 KiB
Python
from sqlalchemy import create_engine
|
|
from sqlalchemy.orm import sessionmaker
|
|
from sqlalchemy.exc import IntegrityError
|
|
|
|
from app.api.cruds.base import create, read, update, delete, read_all
|
|
from app.schema.library import Library
|
|
from ..db.config.config import get_engine_configuration
|
|
|
|
from sqlalchemy.exc import NoResultFound
|
|
|
|
from .exceptions import LibraryCreationException
|
|
from .exceptions import LibraryReadException
|
|
from .exceptions import LibraryUpdateException
|
|
|
|
import logging
|
|
logging.basicConfig(level=logging.DEBUG)
|
|
logger = logging.getLogger(__name__)
|
|
|
|
class LibraryController:
|
|
def __init__(self, library_id=None, *, engine_string=None, echo=False, engine=None):
|
|
if engine_string is None:
|
|
engine_string, echo = get_engine_configuration()
|
|
if engine is None:
|
|
self._engine = create_engine(engine_string, echo=echo=="true", future=True)
|
|
else:
|
|
self._engine = engine
|
|
self._Session = sessionmaker(bind=self.engine)
|
|
self._session = self._Session()
|
|
self._library = None
|
|
if library_id is not None:
|
|
self.read(library_id)
|
|
|
|
self._libraries = []
|
|
|
|
|
|
def __del__(self):
|
|
self.session.close()
|
|
|
|
def __enter__(self):
|
|
return self
|
|
|
|
def __exit__(self, *exc):
|
|
del(self)
|
|
return False
|
|
|
|
@property
|
|
def data(self):
|
|
return self._library
|
|
|
|
@property
|
|
def engine(self):
|
|
return self._engine
|
|
|
|
@property
|
|
def Session(self):
|
|
return self._Session
|
|
|
|
@property
|
|
def session(self):
|
|
return self._session
|
|
|
|
@property
|
|
def libraries(self):
|
|
if self._library and len(self._libraries) == 0:
|
|
self._libraries = [self._library]
|
|
return self._libraries
|
|
|
|
def set_library(self, _id):
|
|
libraries = filter(lambda x: x.get("id") == _id, self.libraries)
|
|
if len(libraries) == 1:
|
|
self._library = libraries[0]
|
|
return True
|
|
return False
|
|
|
|
#CRUDS
|
|
def create(self, library:Library):
|
|
try:
|
|
self._library = create(self.session, library)
|
|
except IntegrityError as e:
|
|
raise LibraryCreationException(
|
|
"Cannot create library",
|
|
f"{e.orig}",
|
|
"library",
|
|
str(library)
|
|
)
|
|
return self
|
|
|
|
def read(self, _id):
|
|
try:
|
|
self._library = read(self.session, _id, Library)
|
|
except NoResultFound as e:
|
|
raise LibraryReadException(
|
|
f"Cannot read Library with id {_id}",
|
|
f"{e}",
|
|
"library",
|
|
_id
|
|
)
|
|
return self
|
|
|
|
def read_all(self):
|
|
self._libraries = read_all(self.session, Library)
|
|
|
|
def update(self):
|
|
try:
|
|
self.session.commit()
|
|
except IntegrityError as e:
|
|
raise LibraryUpdateException(
|
|
f"Cannot update Library",
|
|
f"{e}",
|
|
"library",
|
|
None
|
|
)
|
|
|
|
def delete(self):
|
|
delete(self.session, self.data)
|
|
del(self)
|
|
|