Bind app to API.

assets/__init__.py

@@ -49,3 +49,4 @@ from .config import config_generator
 """ App config and resources. """
 from .app_config import app_config
 from .app_resources import app_resources
+

assets/app_route.py

@@ -9,7 +9,7 @@ class app_route:
     def _fail(self, cause: str, **kwargs) -> dict:
         kwargs["cause"] = cause
 
-        return self.__response("fail", cause)
+        return self.__response("fail", **kwargs)
 
     def _require(self, sended: dict, *args) -> None:
         for count in args:

assets/product_app.py

@@ -23,42 +23,74 @@ class product_app(app_route_database):
         self.__users_collection = users
 
     def all(self) -> dict:
-        with self.__products_database as loader:
-            return self.__collection(loader.load_all())
+        try:
+            with self.__products_database as loader:
+                return self.__collection(loader.load_all())
 
+        except Exception as error:
+            return self._fail(str(error))
+    
     def get_barcode(self, target: str) -> dict:
-        with self.__products_database as loader:
-            return self.__single(loader.get_by_barcode(target))
+        try:
+            with self.__products_database as loader:
+                return self.__single(loader.get_by_barcode(target))
+
+        except Exception as error:
+            return self._fail(str(error))
 
     def get_name(self, target: str) -> dict:
-        with self.__products_database as loader:
-            return self.__single(loader.get_by_name(target))
+        try:
+            with self.__products_database as loader:
+                return self.__single(loader.get_by_name(target))
+
+        except Exception as error:
+            return self._fail(str(error))
 
     def search_name(self, target: str) -> dict:
-        with self.__products_database as loader:
-            return self.__collection(loader.search_by_name(target))
+        try:
+            with self.__products_database as loader:
+                return self.__collection(loader.search_by_name(target))
+
+        except Exception as error:
+            return self._fail(str(error))
 
     def search_author(self, target: str) -> dict:
-        with self.__products_database as loader:
-            return self.__collection(loader.search_by_author(target))
+        try:
+            with self.__products_database as loader:
+                return self.__collection(loader.search_by_author(target))
+
+        except Exception as error:
+            return self._fail(str(error))
 
     def check_barcode(self, target: str) -> dict:
-        with self.__products_database as loader:
-            return self.__exists(loader.barcode_in_use(target))
+        try:
+            with self.__products_database as loader:
+                return self.__exists(loader.barcode_in_use(target))
+
+        except Exception as error:
+            return self._fail(str(error))
 
     def check_name(self, target: str) -> dict:
-        with self.__products_database as loader:
-            return self.__exists(loader.name_in_use(target))
+        try:
+            with self.__products_database as loader:
+                return self.__exists(loader.name_in_use(target))
+
+        except Exception as error:
+            return self._fail(str(error))
 
     def create(self, send: dict) -> dict:
-        if not self.__logged_in(send):
-            raise access_denied_exception()
+        try:
+            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)
 
-        with self.__products_database as loader:
-            target = product_builder().modify(send).result
-            result = loader.store(target)
+                return self.__modify(result, "Can nod create product.")
 
-            return self.__modify(result, "Can nod create product.")
+        except Exception as error:
+            return self._fail(str(error))
 
     def __logged_in(self, send: dict) -> bool:
         if not "apikey" in send:
@@ -66,7 +98,11 @@ class product_app(app_route_database):
 
         return self.__users.get(send["apikey"]) is not None
 
-    def __select_by_sended(self, send: dict) -> product | None:
+    def __select_by_sended(
+        self, 
+        send: dict, 
+        loader: product_loader
+    ) -> product | None:
         barcode = None
         name = None
 
@@ -82,45 +118,50 @@ class product_app(app_route_database):
 
         if barcode is None and name is None:
             content = "Give target_barcode or target_name"
-            raise bar_request_exception(content)
+            raise bad_request_exception(content)
 
-        with self.__product_database as laoder:
-            result = None
+        result = None
 
-            if barcode is not None:
-                result = loader.get_by_barcode(barcode)
+        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()
+        if name is not None:
+            result = loader.get_by_name(name)
+        
+        if result is None:
+            raise not_found_exception()
 
-            return result
+        return result
 
     def update(self, send: dict) -> dict:
-        if not self.__logged_in(send):
-            raise access_denied_exception()
+        try:
+            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)
+            with self.__products_database as loader:
+                target = self.__select_by_sended(send, loader)
+                updated = product_builder(target).modify(send).result
+                result = loader.store(updated)
 
-            return self.__modify(result, "Can not update product.")
+                return self.__modify(result, "Can not update product.")
 
-    def delete(self, send: dict) -> dict:
-        if not self.__logged_in(send):
-            raise access_denied_exception()
+        except Exception as error:
+            return self._fail(str(error))
 
-        target = self.__select_by_sended(send)
+    def delete(self, send: dict) -> dict:
+        try:
+            if not self.__logged_in(send):
+                raise access_denied_exception()
+
+            with self.__products_database as loader:
+                target = self.__select_by_sended(send, loader)
+                result = loader.drop(target)
+                
+                return self.__modify(result, "Can not delete product.")
+        
+        except Exception as error:
+            return self._fail(str(error))
 
-        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()

assets/product_loader.py

@@ -9,6 +9,7 @@ from .validator import name_validator
 from .validator import description_validator
 from .validator import author_validator
 from .exception import validator_exception
+from .exception import exists_exception
 
 class product_loader(sqlmodel.Session):
     """
@@ -158,10 +159,10 @@ class product_loader(sqlmodel.Session):
         """
 
         if self.barcode_in_use(target.barcode):
-            exists_exception("barcode")
+            raise exists_exception("barcode")
         
         if self.name_in_use(target.name):
-            exists_exception("name")
+            raise exists_exception("name")
 
         self.add(target)
         self.commit()
@@ -184,7 +185,7 @@ class product_loader(sqlmodel.Session):
         """
 
         if not target.ready:
-            return False
+            raise RuntimeError("Target is not ready yet.")
 
         if not target.in_database:
             return self.__append(target)
@@ -192,22 +193,22 @@ class product_loader(sqlmodel.Session):
         name_use = self.get_by_name(target.name)
 
         if name_use is not None and name_use.id != target.id:
-            return False
+            raise exists_exception("name")
 
         barcode_use = self.get_by_barcode(target.barcode)
 
         if barcode_use is not None and barcode_use.id != target.id:
-            return False
+            raise exists_exception("barcode")
 
         try:
-            self.add(target)
+            target.sqlmodel_update(target.model_dump(exclude = "id"))
             self.commit()
             self.refresh(target)
 
             return True
-        
+       
         except:
-            return False
+           return False
 
     def drop(self, target: product) -> bool:
         """
@@ -237,4 +238,4 @@ class product_loader(sqlmodel.Session):
     def __select(self) -> sqlmodel.sql._expression_select_cls.SelectOfScalar:
         """ New product selector. """
         
-        return sqlmodel.select(product)
+        return sqlmodel.select(product)

assets/requests.py

@@ -0,0 +1,69 @@
+import pydantic
+
+class user_login_request(pydantic.BaseModel):
+    nick: str
+    password: str
+
+    model_config = {
+        "json_schema_extra": {
+            "excamples": [
+                {
+                    "nick": "test",
+                    "password": "QWERTYZ"
+}
+            ]
+        }
+    }
+
+class user_get_request(pydantic.BaseModel):
+    apikey: str
+    
+    model_config = {
+        "json_schema_extra": {
+            "excamples": [
+                {
+                    "apikey": "af...699",
+                }
+            ]
+        }
+    }
+
+class product_request(pydantic.BaseModel):
+    apikey: str
+    name: str
+    description: str
+    author: str
+    image: str
+    stock_count: int
+    barcode: str
+    
+    model_config = {
+        "json_schema_extra": {
+            "excamples": [
+                {
+                    "apikey": "af...69",
+                    "name": "Product Name",
+                    "description": "Product description.",
+                    "author": "Product author.",
+                    "image": "https://api.com/image.png",
+                    "stocik_count": 10,
+                    "barcode": "509282819938"
+                }
+            ]
+        }
+    }
+
+class apikey_request(pydantic.BaseModel):
+    apikey: str
+    
+    model_config = {
+        "json_schema_extra": {
+            "excamples": [
+                {
+                    "apikey": "af...699",
+                }
+            ]
+        }
+    }
+
+

assets/server.py

@@ -0,0 +1,138 @@
+import fastapi
+import fastapi.staticfiles
+import fastapi.responses
+import pathlib
+import os
+
+from . import *
+from .requests import *
+
+class server(fastapi.FastAPI):
+    def __init__(self, config_file: str) -> None:
+        super().__init__()
+
+        location = pathlib.Path(config_file)
+        
+        self.__resources = config_loader(app_config) \
+        .load(location) \
+        .resources
+        
+        self.__static()
+        self.__route_product()
+        self.__route_users()
+
+    @property
+    def resources(self) -> app_resources:
+        return self.__resources
+
+    @property
+    def product_app(self) -> product_app:
+        return self.__resources.product_app
+
+    @property
+    def users_app(self) -> users_app:
+        return self.__resources.users_app
+
+    def __static(self) -> None:
+        @self.get("/")
+        async def root() -> str:
+            with pathlib.Path("static/core.html").open() as core:
+                return fastapi.responses.HTMLResponse(
+                    content = core.read(),
+                    status_code = 200
+                )
+    
+        directory = fastapi.staticfiles.StaticFiles(
+            directory = pathlib.Path("static/")
+        )
+
+        self.mount(
+            "/app", 
+            directory, 
+            name = "app_frontend"
+        )
+    
+    def __route_product(self) -> None:
+        @self.get("/products")
+        async def products() -> dict:
+            return self.product_app.all()
+
+        @self.get("/product/get/barcode/{barcode}")
+        async def product_get_barcode(barcode: int) -> dict:
+            return self.product_app.get_barcode(str(barcode))
+
+        @self.get("/product/get/name/{name}")
+        async def product_get_name(name: str) -> dict:
+            return self.product_app.get_name(name)
+
+        @self.get("/product/search/name/{name}")
+        async def product_search_name(name: str) -> dict:
+            return self.product_app.search_name(name)
+
+        @self.get("/product/search/author/{author}")
+        async def product_search_author(author: str) -> dict:
+            return self.product_app.search_author(author)
+
+        @self.get("/product/check/barcode/{barcode}")
+        async def product_check_barcode(barcode: int) -> dict:
+            return self.product_app.check_barcode(str(barcode))
+
+        @self.get("/product/check/name/{name}")
+        async def product_check_name(name: str) -> dict:
+            return self.product_app.check_name(name)
+
+        @self.post("/product/update/barcode/{barcode}")
+        async def product_barcode_update(
+            barcode: str, 
+            product: product_request
+        ) -> dict:
+            body = product.dict()
+            body["target_barcode"] = barcode
+
+            return self.product_app.update(body)
+
+        @self.post("/product/update/name/{name}")
+        async def product_name_update(
+            name: str, 
+            product: product_request
+        ) -> dict:
+            body = product.dict()
+            body["target_name"] = name
+
+            return self.product_app.update(body)
+
+        @self.delete("/product/barcode/{barcode}")
+        async def product_barcode_delete(
+            barcode: str,
+            request: apikey_request
+        ) -> dict:
+            body = request.dict()
+            body["target_barcode"] = barcode
+
+            return self.product_app.delete(body)
+
+        @self.delete("/product/name/{name}")
+        async def product_name_delete(
+            name: str,
+            request: apikey_request
+        ) -> dict:
+            body = request.dict()
+            body["target_name"] = name
+
+            return self.product_app.delete(body)
+
+        @self.post("/product/create")
+        async def product_create(product: product_request) -> dict:
+            return self.product_app.create(product.dict())
+
+    def __route_users(self) -> None:    
+        @self.post("/user/login")
+        async def user_login(user: user_login_request) -> dict:
+            return self.users_app.login(user.nick, user.password)
+        
+        @self.post("/user")
+        async def user_get(body: user_get_request) -> dict:
+            return self.users_app.get(body.apikey)
+        
+
+instance = server(os.environ["config_file"])

assets/users_app.py

@@ -8,11 +8,11 @@ class users_app(app_route):
     def __init__(self, collection: users_collection) -> None:
         self.__collection = collection
 
-    def login(self, sended: dict) -> dict:
-        self._require(sended, "nick", "password")
-        
-        nick = sended["nick"]
-        password = sended["password"]
+    def login(
+        self, 
+        nick: str,
+        password: str
+    ) -> dict:
         target = self.collection.login(nick, password)
 
         if target is None:
@@ -20,10 +20,7 @@ class users_app(app_route):
 
         return self._success(apikey = target.apikey)
 
-    def get(self, sended: dict) -> dict:
-        self._require(sended, "apikey")
-
-        apikey = sended["apikey"]
+    def get(self, apikey: str) -> dict:
         target = self.collection.get(apikey)
 
         if target is None:

core.py

@@ -3,6 +3,8 @@ import enum
 import pathlib
 import getpass
 import json
+import os
+import uvicorn
 
 from assets import *
 
@@ -48,9 +50,21 @@ def server(
     """
     Start app on selected port and interfaces.
     """
-
-    print(port)
-    print(address)
+    os.environ["config_file"] = str(config)
+    server_config = uvicorn.Config(
+        "assets.server:instance", 
+        port = port, 
+        host = address, 
+        log_level = "info"
+    )
+
+    app = uvicorn.Server(server_config)
+    
+    try:    
+        app.run()
+    except json.JSONDecodeError as error:
+        print("Can not parse config and user database.")
+        print(str(error))
 
 @app.command()
 def user(

requirements.txt

@@ -1,3 +1,4 @@
 sqlmodel
 fastapi
 typer-slim
+uvicorn

static/core.html

@@ -0,0 +1,5 @@
+<!DOCTYPE html>
+
+<html>
+
+</html>

tests/003-product_app.py

@@ -20,7 +20,15 @@ drop_database()
 connection = sqlmodel.create_engine("sqlite:///003-product_app.db")
 sqlmodel.SQLModel.metadata.create_all(connection)
 
-app = assets.product_app(connection)
+users = assets.users_collection()
+
+factory = assets.user_factory()
+factory.nick = "test"
+factory.password = "12345678"
+
+users.add(factory.result)
+
+app = assets.product_app(connection, users)
 
 print("App initialized.")
 
@@ -30,7 +38,8 @@ create = app.create({
     "description": "This is sample name.",
     "author": "John Snow",
     "image": "https://uuu.owo.pl",
-    "stock_count": "10"
+    "stock_count": "10",
+    "apikey": factory.apikey
 })
 
 print("Create:")
@@ -55,7 +64,8 @@ create = app.create({
     "description": "This is sample item second.",
     "author": "other",
     "image": "https://test.pl",
-    "stock_count": "20"
+    "stock_count": "20",
+    "apikey": factory.apikey
 })
 
 print("Create second:")
@@ -68,4 +78,19 @@ print("Select all:")
 print(alls)
 print()
 
+modify = app.update({
+    "barcode": "210987654321",
+    "target_barcode": "210987654321",
+    "name": "Second sample",
+    "description": "This is sample item second.",
+    "author": "other",
+    "image": "https://test.pl",
+    "stock_count": "20",
+    "apikey": factory.apikey
+})
+
+print("Modify:")
+print(modify)
+print()
+
 drop_database()

tests/006-resources.py

@@ -1,19 +1 @@
-import pathlib
-
-current = pathlib.Path(__file__).parent
-root = current.parent
-
-import sys
-sys.path.append(str(root))
-
-import assets
-
-config = pathlib.Path("./006-resources.py")
-
-if config.is_file():
-    config.unlink()
-
-assets.config_generator(assets.app_config).save(config)
-app = assets.config_loader(assets.app_config).load(config).resources
-
-config.unlink()
+{"database_uri": "sqlite:///database.db", "users_file": "users.json"}