Continue working...

assets/__init__.py

@@ -29,6 +29,7 @@ from .app_route import app_route
 from .app_route import app_route_database
 from .users_app import users_app
 from .product_app import product_app
+from .deep_request import deep_request
 
 """ User and user helpers. """
 from .user import user
@@ -53,3 +54,10 @@ from .app_resources import app_resources
 """ Image managment. """
 from .image import image
 from .directory_image import directory_image
+
+""" Reservation managment. """
+from .reservation import reservation
+from .reservation import reservation_factory
+from .reservations_collection import reservations_collection
+from .reservation_loader import reservation_loader
+from .reservation_app import reservation_app

assets/app_resources.py

@@ -8,6 +8,7 @@ from .users_collection import users_collection
 from .product_app import product_app
 from .users_app import users_app
 from .directory_image import directory_image
+from .reservation_app import reservation_app
 
 class app_resources:
     def __init__(self, config: object) -> None:
@@ -31,10 +32,19 @@ class app_resources:
             self.images
         )
 
+        self.__reservation_app = reservation_app(
+            self.database,
+            self.users
+        )
+
     @property
     def users_app(self) -> users_app:
         return self.__users_app
 
+    @property
+    def reservation_app(self) -> reservation_app:
+        return self.__reservation_app
+
     @property
     def images(self) -> directory_image:
         return self.__directory_image

assets/deep_request.py

@@ -0,0 +1,39 @@
+class deep_request:
+    """
+    This class is responsible for processing requests betwen 
+    fastapi endpoints and apps.
+    """
+
+    def __init__(self, params: dict) -> None:
+        """
+        This create new request.
+        
+        Params:
+            params (dict): Request dict from endpoint
+        """
+
+        self.__params = params.copy()
+
+    @property
+    def _params(self) -> dict:  
+        """ This return params for child class. """
+
+        return self.__params
+
+    def __getattr__(self, key: str) -> None | str | int | float:
+        """ 
+        This trying to load key from request. When key could not being 
+        found, then return None.
+
+        Params:
+            key (str): Key to load from request
+        
+        Returns:
+            (None | str | int | float): Return key from request
+        """
+
+        if not key in self._params:
+            return None
+        
+        return self._params[key]
+

assets/exception.py

@@ -57,3 +57,10 @@ class reservation_exception(Exception):
 
 class reservation_loader_exception(Exception):
     pass
+
+class database_exception(Exception):
+    def __init__(self) -> None:
+        super().__init__("Database can not commit changes.")
+
+class incorrect_target_exception(Exception):
+    pass

assets/product.py

@@ -41,6 +41,23 @@ class product(sqlmodel.SQLModel, table = True):
         unique = True
     )
 
+    reservations: list["reservation"] = sqlmodel.Relationship(
+        back_populates = "target"
+    )
+
+    @property
+    def on_stock(self) -> int:
+        current = self.stock_count - len(self.reservations)
+        
+        if current < 0:
+            return 0
+
+        return current
+    
+    @property
+    def avairable(self) -> bool:
+        return self.on_stock > 0
+
     def save_copy(self) -> object:
         """
         This return clone of the product, which is not connect with 

assets/product_loader.py

@@ -226,6 +226,9 @@ class product_loader(sqlmodel.Session):
             return False
 
         try:
+            for count in target.reservations:
+                self.delete(count)
+
             self.delete(target)
             self.commit()
 

assets/requests.py

@@ -1,3 +1,4 @@
+import typing
 import pydantic
 
 class user_login_request(pydantic.BaseModel):
@@ -10,7 +11,29 @@ class user_login_request(pydantic.BaseModel):
                 {
                     "nick": "test",
                     "password": "QWERTYZ"
-}
+                }
+            ]
+        }
+    }
+
+class reservation_request(pydantic.BaseModel):
+    apikey: str
+    target_name: typing.Optional[str]
+    target_barcode: typing.Optional[str]
+    email: typing.Optional[str]
+    phone_number: typing.Optional[str]
+
+    model_config = {
+        "json_schema_extra": {
+            "excamples": [
+                {
+                    "target_barcode": "123456789012",
+                    "email": "sample@noreply"
+                },
+                {
+                    "target_name": "Sample",
+                    "phone_number": "+48 123456789"
+                }
             ]
         }
     }

assets/reservation.py

@@ -28,8 +28,8 @@ class reservation(sqlmodel.SQLModel, table = True):
         unique = False
     ) 
 
-    target: product | None = sqlmodel.Relationshio(
-        back_populates = "product"
+    target: product | None = sqlmodel.Relationship(
+        back_populates = "reservations"
     )
 
     @property 
@@ -63,7 +63,7 @@ class reservation(sqlmodel.SQLModel, table = True):
         if self.target is None:
             return content + "Target: not set"
 
-        return content + "Target: #" + str(self.target) + "\n"
+        return content + "Target: #" + str(self.target.id) + "\n"
     
 class reservation_factory:
     def __init__(self, target: reservation | None = None) -> None:
@@ -73,10 +73,10 @@ class reservation_factory:
             self.__target = reservation()
 
     def target(self, item: product) -> object:
-        if item.in_database:
+        if not item.in_database:
             raise reservation_exception("Target item not in database.")
 
-        self.__target.target = item.id
+        self.__target.target = item
         return self
 
     def phone_number(self, number: str) -> object:
@@ -93,7 +93,7 @@ class reservation_factory:
         self.__target.email = email
         return self
     
-    def result() -> reservation:
+    def result(self) -> reservation:
         return self.__target
 
         

assets/reservation_app.py

@@ -0,0 +1,235 @@
+import typing
+import sqlalchemy
+import sqlalchemy.engine.base
+
+from .app_route import app_route_database
+from .product import product 
+from .product_loader import product_loader
+from .exception import bad_request_exception
+from .exception import not_found_exception
+from .exception import access_denied_exception
+from .exception import database_exception
+from .exception import incorrect_target_exception
+from .users_collection import users_collection
+from .reservation import reservation 
+from .reservation import reservation_factory
+from .reservation_loader import reservation_loader
+from .deep_request import deep_request
+from .reservations_collection import reservations_collection
+from .reservations_response import reservations_response
+
+class reservation_deep_request(deep_request):
+    @property
+    def has_product(self) -> bool:  
+        return self.target_barcode is not None or self.target_name is not None
+
+    @property 
+    def has_email(self) -> bool:
+        return self.email is not None
+
+    @property
+    def has_phone_number(self) -> bool:
+        return self.phone_number is not None
+
+    @property
+    def has_apikey(self) -> bool:
+        return self.apikey is not None
+
+class reservation_app(app_route_database):
+    def __init__(
+        self,
+        connection: sqlalchemy.engine.base.Engine,
+        users: users_collection
+    ) -> None:
+        super().__init__(connection)
+        self.__users_collection = users
+
+    def get_by_user(self, params: dict) -> dict:
+        request = reservation_deep_request(params)
+        
+        try:
+            with self.__reservation_loader as loader:
+                collection = self.__load_by_user(request, loader)
+                self.__reservation_user_filter(collection, request)
+
+                reservations_list = reservations_response() \
+                    .collection(collection.results())
+
+                return self._success(reservations = reservations_list)
+        
+        except Exception as error:
+            return self.__catch(error)
+
+    def get_by_product(self, params: dict) -> dict:
+        request = reservation_deep_request(params)
+
+        try:
+            self.__login(request)
+            item = self.__load_product(request)
+            
+            with self.__reservation_loader as loader:
+                collection = loader.get_by_target(item) 
+                self.__reservation_user_filter(collection, request)
+ 
+                reservations_list = reservations_response() \
+                    .collection(collection.results())
+
+                return self._success(reservations = reservations_list)
+        
+        except Exception as error:
+            return self.__catch(error)
+
+    def rent_product(self, params: dict) -> dict:
+        request = reservation_deep_request(params)
+        
+        try:
+            self.__login(request)
+            item = self.__load_product(request)
+            rent = self.__create_factory(request).target(item).result()
+            
+            with self.__reservation_loader as loader:
+                if not loader.store(rent):
+                    raise database_exception()
+
+            return self._success()
+
+        except Exception as error:
+            return self.__catch(error)
+
+    def give_back_product(self, params: dict) -> dict:
+        request = reservation_deep_request(params)
+
+        try:
+            self.__login(request)
+            target = self.__get_reservation(request)
+
+            with self.__reservation_loader as loader:
+                if loader.drop(target):
+                    return self._success()
+                
+                raise database_exception()
+        
+        except Exception as error:
+            return self.__catch(error)
+
+    def __reservation_user_filter(
+        self, 
+        collection: reservations_collection, 
+        request: deep_request
+    ) -> bool:
+        if not request.has_email and not request.has_phone_number:
+            return False
+
+        if request.has_email:
+            collection.by_email(request.email)
+
+        if request.has_phone_number:
+            collection.by_phone_number(request.phone_number)
+
+        return True
+
+    def __reservation_product_filter(
+        self,
+        collection: reservations_collection,
+        request: deep_request
+    ) -> bool:
+        if not request.has_product:
+            return False
+
+        item = self.__load_product(request)
+        collection.by_target(item)
+
+        return True
+
+    def __get_reservation(self, request: deep_request) -> reservation:
+        with self.__reservation_loader as loader:
+            item = self.__load_product(request)
+            collection = loader.get_by_target(item)
+            
+            if not self.__reservation_user_filter(collection, request):
+                raise bad_request_exception("Not contaim email or phone")
+
+            result = collection.results()
+
+            if len(result) == 0:    
+                raise not_found_exception()
+            
+            if len(result) > 1:
+                raise incorrect_target_exception("Found too much targets.")
+
+            return result[0]
+
+    def __load_by_user(
+        self, 
+        request: deep_request,
+        loader: reservation_loader
+    ) -> reservations_collection:
+        if request.has_email:
+            return loader.get_by_email(request.email)
+
+        if request.has_phone_number:
+            return loader.get_by_phone_number(request.phone_number)
+
+        raise bad_request_exception("Has no email or phone number")
+
+    def __create_factory(self, request: deep_request) -> reservation_factory:
+        if not request.has_email and not request.has_phone_number:
+            raise bad_request_exception("Not contain email or/and phone")
+
+        factory = reservation_factory()
+
+        if request.has_email:
+            factory.email(request.email)
+
+        if request.has_phone_number:
+            factory.phone_number(request.phone_number)
+
+        return factory
+
+    def __login(self, request: deep_request) -> None:
+        if not request.has_apikey:
+            raise access_denied_exception()
+
+        if self.__users_collection.get(request.apikey):
+            return 
+
+        raise access_denied_exception()
+
+    def __load_product(self, request: deep_request) -> product:  
+        if request.target_barcode is not None:
+            return self.__product_by_barcode(request)
+
+        if request.target_name is not None:
+            return self.__product_by_name(request)
+
+        raise bad_request_exception("Not contain barcode or name")
+
+    def __product_by_barcode(self, request: deep_request) -> product:
+        with self.__product_loader as loader:
+            product = loader.get_by_barcode(request.target_barcode)
+
+            if product is not None:
+                return product
+
+            raise not_found_exception()
+
+    def __product_by_name(self, request: deep_request) -> product:
+        with self.__product_loader as loader:
+            product = loader.get_by_name(request.target_name) 
+
+            if product is not None:
+                return product
+            
+            raise not_found_exception()
+
+    def __catch(self, error: Exception) -> dict:
+        return self._fail(str(error))
+
+    @property
+    def __reservation_loader(self):
+        return reservation_loader(self._connection)
+
+    @property
+    def __product_loader(self):
+        return product_loader(self._connection)
+        

assets/reservation_loader.py

@@ -8,13 +8,26 @@ from .reservations_collection import reservations_collection
 
 class reservation_loader(sqlmodel.Session): 
     def get_by_email(self, target: str) -> reservations_collection:
-        pass
-
+        return self.__get(reservation.email == target)
+    
     def get_by_phone_number(self, target: str) -> reservations_collection:
-        pass
+        return self.__get(reservation.phone_number == target)
 
     def get_by_target(self, target: product) -> reservations_collection:
-        pass
+        if not target.in_database:
+            return reservations_collection()
+
+        return self.__get(reservation.target == target)
+
+    def __get(self, condition: object) -> reservations_collection:
+        query = self.__select.where(condition)
+        collection = reservations_collection()
+        result = self.exec(query)
+
+        for item in result:
+            collection.append(item)
+
+        return collection
 
     def store(self, target: reservation) -> bool:
         if not target.ready:

assets/reservations_collection.py

@@ -1,3 +1,5 @@
+import typing
+
 from .product import product
 from .reservation import reservation
 
@@ -9,7 +11,7 @@ class reservations_collection:
         self.__content.append(target)
         return self
 
-    def _filter(self, action: function) -> object:
+    def _filter(self, action: typing.Callable) -> object:
         new = list()
 
         for item in self.__content:
@@ -31,6 +33,5 @@ class reservations_collection:
     def by_phone_number(self, target: str) -> object:
         return self._filter(lambda item: item.phone_number == target)
     
-    @property
-    def result(self) -> list:
+    def results(self) -> list:
         return self.__content

assets/reservations_response.py

@@ -0,0 +1,28 @@
+from .reservation import reservation
+from .reservations_collection import reservations_collection
+
+class reservations_response:
+    def collection(self, targets: list) -> list:
+        result = list()
+
+        for count in targets:
+            if not isinstance(count, reservation):
+                raise TypeError("Reservations list not contain reservation.")
+
+            result.append(self.single(count))
+
+        return result
+
+    def single(self, target: reservation) -> dict:
+        response = {
+            "target_name": target.target.name,
+            "target_barcode": target.target.barcode
+        }
+        
+        if target.email is not None:
+            response["email"] = target.email
+
+        if target.phone_number is not None:
+            response["phone_number"] = target.phone_number
+
+        return response

assets/server.py

@@ -20,6 +20,7 @@ class server(fastapi.FastAPI):
         self.__static()
         self.__route_product()
         self.__route_users()
+        self.__route_reservations()
 
     @property
     def resources(self) -> app_resources:
@@ -33,6 +34,10 @@ class server(fastapi.FastAPI):
     def users_app(self) -> users_app:
         return self.__resources.users_app
 
+    @property
+    def reservation_app(self) -> reservation_app:
+        return self.__resources.reservation_app
+
     def __static(self) -> None:
         @self.get("/")
         async def root() -> str:
@@ -61,7 +66,88 @@ class server(fastapi.FastAPI):
             covers_directory,
             name = "covers"
         )
-    
+
+    def __route_reservations(self) -> None:
+        @self.post("/reservations/user/email/{email}")
+        async def reservations_email(
+            email: str,
+            reservation: reservation_request
+        ) -> dict:
+            body = reservation.dict()
+            body["email"] = email
+
+            return self.reservation_app.get_by_user(body)
+
+        @self.post("/reservations/user/phone_number/{phone_number}")
+        async def reservations_email(
+            phone_number: str,
+            reservation: reservation_request
+        ) -> dict:
+            body = reservation.dict()
+            body["phone_number"] = phone_number
+
+            return self.reservation_app.get_by_user(body)
+
+        @self.post("/reservations/product/barcode/{barcode}")
+        async def reservations_barcode(
+            barcode: int,
+            reservation: reservation_request
+        ) -> dict:
+            body = reservation.dict()
+            body["target_barcode"] = barcode
+
+            return self.reservation_app.get_by_product(body)
+
+        @self.post("/reservations/product/name/{name}")
+        async def reservations_name(
+            name: str,
+            reservation: reservation_request
+        ) -> dict:
+            body = reservation.dict()
+            body["target_name"] = name
+
+            return self.reservation_app.get_by_product(body)
+
+        @self.post("/rent/product/barcode/{barcode}")
+        async def rent_barcode(
+            barcode: int,
+            reservation: reservation_request
+        ) -> dict:
+            body = reservation.dict()
+            body["target_barcode"] = barcode
+            
+            return self.reservation_app.rent_product(body)
+
+        @self.post("/rent/product/name/{name}")
+        async def rent_name(
+            name: str,
+            reservation: reservation_request
+        ) -> dict:
+            body = reservation.dict()
+            body["target_name"] = name
+            
+            return self.reservation_app.rent_product(body)
+
+        @self.post("/give_back/product/barcode/{barcode}")
+        async def give_back_barcode(
+            barcode: int,
+            reservation: reservation_request
+        ) -> dict:
+            body = reservation.dict()
+            body["target_barcode"] = barcode
+
+            return self.reservation_app.give_back_product(body)
+
+        @self.post("/give_back/product/name/{name}")
+        async def give_back_barcode(
+            name: str,
+            reservation: reservation_request
+        ) -> dict:
+            body = reservation.dict()
+            body["target_name"] = name
+
+            return self.reservation_app.give_back_product(body)
+
     def __route_product(self) -> None:
         @self.get("/products")
         async def products() -> dict:

tests/009-reservation.py

@@ -0,0 +1,135 @@
+import pathlib
+
+current = pathlib.Path(__file__).parent
+root = current.parent
+
+import sys
+sys.path.append(str(root))
+
+import assets
+import sqlmodel
+
+def drop_database() -> None:
+    db = pathlib.Path("./009-reservation.db")
+
+    if db.is_file():
+        db.unlink()
+
+drop_database()
+
+connection = sqlmodel.create_engine("sqlite:///009-reservation.db")
+sqlmodel.SQLModel.metadata.create_all(connection)
+
+builder = assets.product_factory()
+builder.name = "Sample name"
+builder.author = "Simple UwU Artist"
+builder.description = "This is simple description"
+builder.barcode = "123456789012"
+builder.stock_count = 10
+
+product = builder.result
+
+print("Create first product:")
+print(product)
+print()
+
+with assets.product_loader(connection) as loader:
+    print("Inserting product...")
+    loader.store(product)
+
+    print("Inserted:")
+    print(product)
+    print()
+
+with assets.product_loader(connection) as loader:
+    print("Loading product to make reservation...")
+    target = loader.get_by_barcode("123456789012")
+    print("Loaded: ")
+    print(target)
+
+    before = target.on_stock
+
+    reservation = assets \
+    .reservation_factory() \
+    .target(target) \
+    .email("[email protected]") \
+    .result()
+
+    print(reservation)
+    print()
+
+with assets.reservation_loader(connection) as loader:
+    print("Inserting it into database...")
+    loader.store(reservation)
+    print(reservation)
+
+with assets.product_loader(connection) as loader:
+    print("Loading product to make reservation...")
+    target = loader.get_by_barcode("123456789012")
+    print("Loaded: ")
+    print(target)
+
+    reservation = assets \
+    .reservation_factory() \
+    .target(target) \
+    .email("[email protected]") \
+    .result()
+
+    print(reservation)
+    print()
+
+with assets.reservation_loader(connection) as loader:
+    print("Inserting it into database...")
+    loader.store(reservation)
+    print(reservation)
+
+with assets.product_loader(connection) as loader:
+    print("Loading product to check that on stock calc work...")
+    target = loader.get_by_barcode("123456789012")
+    after = target.on_stock
+    
+    print("Before: " + str(before))
+    print("After: " + str(after))
+    print()
+
+    before = after
+
+with assets.reservation_loader(connection) as loader:
+    print("Loading reservation for product...")
+    
+    collection = loader \
+    .get_by_target(target) \
+    .by_email("[email protected]") \
+    .results()
+    
+    if len(collection) != 1:
+        print("Fail, collection:")
+        print(collection)
+        raise Exception("Collection is not propertly filtered.")
+    
+    reservation = collection[0]
+
+    print("Loaded.")
+    print(reservation)
+    print()
+    
+    print("Removing it from database...")
+    loader.drop(reservation)
+    print("Removed.")
+    print()
+
+with assets.product_loader(connection) as loader:
+    print("Loading product to check that on stock calc work...")
+    target = loader.get_by_barcode("123456789012")
+    after = target.on_stock
+    
+    print("Before: " + str(before))
+    print("After: " + str(after))
+    print()
+
+    print("Trying to drop product...")
+    loader.drop(target)
+    print("Removed.")
+
+drop_database()
+

tests/010-deep_request.py

@@ -0,0 +1,39 @@
+import pathlib
+
+current = pathlib.Path(__file__).parent
+root = current.parent
+
+import sys
+sys.path.append(str(root))
+
+import assets
+
+class deep(assets.deep_request):
+    @property
+    def has_x(self):
+        return self.x is not None
+           
+print("Testing.")
+test = deep({
+    "a": 10,
+    "b": 20
+})
+
+print("Test.a: " + str(test.a))
+print("Test.b: " + str(test.b))
+print("Test.x: " + str(test.x))
+print("Test.has_x: " + str(test.has_x))
+print()
+
+print("Testing.")
+test = deep({
+    "a": 10,
+    "b": 20,
+    "x": 30
+})
+
+print("Test.a: " + str(test.a))
+print("Test.b: " + str(test.b))
+print("Test.x: " + str(test.x))
+print("Test.has_x: " + str(test.has_x))
+print()

tests/011-reservation_app.py

@@ -0,0 +1,142 @@
+import pathlib
+
+current = pathlib.Path(__file__).parent
+root = current.parent
+
+import sys
+sys.path.append(str(root))
+
+import assets
+import sqlmodel
+
+def drop_database() -> None:
+    db = pathlib.Path("./009-reservation.db")
+
+    if db.is_file():
+        db.unlink()
+
+drop_database()
+
+connection = sqlmodel.create_engine("sqlite:///009-reservation.db")
+sqlmodel.SQLModel.metadata.create_all(connection)
+
+builder = assets.product_factory()
+builder.name = "Sample name"
+builder.author = "Simple UwU Artist"
+builder.description = "This is simple description"
+builder.barcode = "123456789012"
+builder.stock_count = 10
+
+product = builder.result
+
+print("Create first product:")
+print(product)
+print()
+
+with assets.product_loader(connection) as loader:
+    print("Inserting product...")
+    loader.store(product)
+
+    print("Inserted:")
+    print(product)
+    print()
+
+with assets.product_loader(connection) as loader:
+    print("Loading product to make reservation...")
+    target = loader.get_by_barcode("123456789012")
+    print("Loaded: ")
+    print(target)
+
+    before = target.on_stock
+
+    reservation = assets \
+    .reservation_factory() \
+    .target(target) \
+    .email("[email protected]") \
+    .result()
+
+    print(reservation)
+    print()
+
+with assets.reservation_loader(connection) as loader:
+    print("Inserting it into database...")
+    loader.store(reservation)
+    print(reservation)
+
+with assets.product_loader(connection) as loader:
+    print("Loading product to make reservation...")
+    target = loader.get_by_barcode("123456789012")
+    print("Loaded: ")
+    print(target)
+
+    reservation = assets \
+    .reservation_factory() \
+    .target(target) \
+    .email("[email protected]") \
+    .result()
+
+    print(reservation)
+    print()
+
+with assets.reservation_loader(connection) as loader:
+    print("Inserting it into database...")
+    loader.store(reservation)
+    print(reservation)
+
+users = assets.users_collection()
+
+factory = assets.user_factory()
+factory.nick = "test"
+factory.password = "12345678"
+
+users.add(factory.result)
+
+app = assets.reservation_app(connection, users)
+
+print("App initialized.")
+print()
+
+print("Loading all reservations:")
+print(app.get_by_product({
+    "apikey": factory.apikey,
+    "target_barcode": "123456789012",
+}))
+print(app.get_by_user({
+    "apikey": factory.apikey,
+    "email": "[email protected]"
+}))
+print()
+
+print("Giving back item:")
+print(app.give_back_product({
+    "apikey": factory.apikey,
+    "target_barcode": "123456789012",
+    "email": "[email protected]"
+}))
+print("After:")
+print(app.get_by_product({
+    "apikey": factory.apikey,
+    "target_barcode": "123456789012",
+}))
+print()
+print("Rent product:")
+print(app.rent_product({
+    "apikey": factory.apikey,
+    "target_barcode": "123456789012",
+    "email": "[email protected]"
+}))
+print(app.rent_product({
+    "apikey": factory.apikey,
+    "target_barcode": "123456789012",
+    "email": "[email protected]",
+    "phone_number": "+48 123456789"
+}))
+print("After:")
+print(app.get_by_product({
+    "apikey": factory.apikey,
+    "target_barcode": "123456789012",
+}))
+print()
+
+drop_database()
+