import typing import sqlalchemy import sqlalchemy.engine.base from .app_route import app_route_database from .product import product from .product import product_factory from .product_loader import product_loader from .product_response import product_response from .product_builder import product_builder from .exception import bad_request_exception from .exception import not_found_exception from .exception import access_denied_exception from .users_collection import users_collection class product_app(app_route_database): def __init__( self, connection: sqlalchemy.engine.base.Engine, users: users_collection ) -> None: super().__init__(connection) self.__users_collection = users def all(self) -> dict: with self.__products_database as loader: return self.__collection(loader.load_all()) def get_barcode(self, target: str) -> dict: with self.__products_database as loader: return self.__single(loader.get_by_barcode(target)) def get_name(self, target: str) -> dict: with self.__products_database as loader: return self.__single(loader.get_by_name(target)) def search_name(self, target: str) -> dict: with self.__products_database as loader: return self.__collection(loader.search_by_name(target)) def search_author(self, target: str) -> dict: with self.__products_database as loader: return self.__collection(loader.search_by_author(target)) def check_barcode(self, target: str) -> dict: with self.__products_database as loader: return self.__exists(loader.barcode_in_use(target)) def check_name(self, target: str) -> dict: with self.__products_database as loader: return self.__exists(loader.name_in_use(target)) def create(self, send: dict) -> dict: if not self.__logged_in(send): raise access_denied_exception() with self.__products_database as loader: target = product_builder().modify(send).result result = loader.store(target) return self.__modify(result, "Can nod create product.") def __logged_in(self, send: dict) -> bool: if not "apikey" in send: return False return self.__users.get(send["apikey"]) is not None def __select_by_sended(self, send: dict) -> product | None: barcode = None name = None if "target_barcode" in send: barcode = send["target_barcode"] if "target_name" in send: name = send["target_name"] if barcode is not None and name is not None: content = "Give only one, target_name or target_barcode" raise bad_request_exception(content) if barcode is None and name is None: content = "Give target_barcode or target_name" raise bar_request_exception(content) with self.__product_database as laoder: result = None if barcode is not None: result = loader.get_by_barcode(barcode) if name is not None: result = loader.get_by_name(name) if result is None: raise not_found_exception() return result def update(self, send: dict) -> dict: if not self.__logged_in(send): raise access_denied_exception() target = self.__select_by_sended(send) updated = product_builder(target).modify(send).result with self.__products_database as loader: result = loader.store(updated) return self.__modify(result, "Can not update product.") def delete(self, send: dict) -> dict: if not self.__logged_in(send): raise access_denied_exception() target = self.__select_by_sended(send) with self.__product_database as loader: result = loader.drop(target) return self.__modify(result, "Can not delete product.") def __modify(self, result: bool, cause: str) -> dict: if result: return self._success() else: return self._fail(cause) def __exists(self, result: bool) -> dict: return self._success(exists = result) def __single(self, target: product) -> dict: if target is None: return self._fail("Can not found product in database.") return self._success(product = product_response(target)) def __collection(self, target: typing.Iterable[product]) -> dict: return self._success(collection = product_response(target)) @property def __users(self) -> users_collection: return self.__users_collection @property def __products_database(self) -> product_loader: return product_loader(self._connection)