[UPD] Exception handling

This commit is contained in:
2026-02-15 20:53:32 +01:00
parent 2a8a44e80b
commit 969333998c
17 changed files with 105 additions and 81 deletions

View File

@@ -6,7 +6,11 @@ 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)
@@ -68,6 +72,7 @@ class LibraryController:
return True
return False
#CRUDS
def create(self, library:Library):
try:
self._library = create(self.session, library)
@@ -76,20 +81,35 @@ class LibraryController:
"Cannot create library",
f"{e.orig}",
"library",
str(library),
400
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)

View File

@@ -1,2 +1,5 @@
from .base import LibraryExceptionBase
from .exc_01000X_data import LibraryDataExection
from .exc_01001X_create import LibraryCreationException
from .exc_01002X_read import LibraryReadException
from .exc_01003X_update import LibraryUpdateException

View File

@@ -1,18 +1,26 @@
class LibraryExceptionBase(Exception):
def __init__(self, code, error):
self.code = code
def __init__(self, name, error, object_name, data, *, status_code=400):
self.code = "000000"
self.name = name
self.error = error
self.object = object_name
self.data = data
self.status_code = status_code
def to_dict(self):
return {
"status": "error",
"name": self.name,
"code": self.code,
"error": self.error,
"status_code": 400
"object": self.object,
"data": self.data,
"status_code": self.status_code
}
def __str__(self):
return f"ERROR {self.code}: {self.error}"
def __repr__(self):
return f"LibraryExceptionBase(code={self.code!r}, error={self.error!r})"
return f"{self.__class__}(code={self.code!r}, error={self.error!r}), " \
f"object={self.object!r}, data={self.data!r}, status_code={self._status_code})"

View File

@@ -0,0 +1,7 @@
from .base import LibraryExceptionBase
#010010
class LibraryDataExection(LibraryExceptionBase):
def __init__(self, name, error, object_name, data, *, status_code=400):
self.code = "010000"
super().__init__(name, error, object_name, data, status_code=status_code)

View File

@@ -2,25 +2,6 @@ from .base import LibraryExceptionBase
#010010
class LibraryCreationException(LibraryExceptionBase):
def __init__(self, name, error, object, data, status_code=400):
def __init__(self, name, error, object_name, data, *, status_code=400):
self.code = "010010"
self.name = name
self.error = error
self.object = object
self.data = data
self.status_code = status_code
def to_dict(self):
return {
"status": "error",
"name": self.name,
"code": self.code,
"error": self.error,
"object": self.object,
"data": self.data,
"status_code": self.status_code
}
def __repr__(self):
return f"LibraryCreationException(code={self.code!r}, error={self.error!r}), " \
f"object={self.object!r}, data={self.data!r}"
super().__init__(name, error, object_name, data, status_code=status_code)

View File

@@ -0,0 +1,7 @@
from .base import LibraryExceptionBase
#010030
class LibraryReadException(LibraryExceptionBase):
def __init__(self, name, error, object_name, data, status_code=404):
super().__init__(name, error, object_name, data, status_code=status_code)
self.code = "010030"

View File

@@ -0,0 +1,7 @@
from .base import LibraryExceptionBase
#010020
class LibraryUpdateException(LibraryExceptionBase):
def __init__(self, name, error, object_name, data, status_code=404):
super().__init__(name, error, object_name, data, status_code=status_code)
self.code = "010020"

View File

@@ -0,0 +1 @@
from .update_item_key import update_item_key

View File

@@ -0,0 +1,21 @@
from ..exceptions import LibraryDataExection
from ...schema.library import Base
def update_item_key(obj:Base, key, value):
if key == "id":
raise LibraryDataExection(
"id is not updatable",
"The key ID is not Updatable",
obj.__class__,
{key: value}
)
try:
obj.__getattribute__(key)
except AttributeError:
raise LibraryDataExection(
f"{key} not in {obj.__class__}",
f"The key {key} is not in {obj.__class__}",
obj.__class__,
{key: value}
)
obj.__setattr__(key, value)

View File

@@ -16,6 +16,7 @@ def handle_exception(e):
"status_code": e.status_code,
"name": e.name,
"error": e.error,
"data": e.data
})
response.content_type = "application/json"
response.status_code = e.status_code

View File

@@ -1,5 +1,4 @@
from flask import request
from sqlalchemy.exc import IntegrityError
from .blueprint import api_library

View File

@@ -12,11 +12,7 @@ logger = logging.getLogger(__name__)
@api_library.route("/<_id>", methods=["DELETE"])
def delete_library(_id):
try: # TODO: function
controller = LibraryController(_id)
except NoResultFound as e:
logger.debug({e})
return { "status": "error", "error": "Library not found" }, 404
controller.delete()

View File

@@ -17,11 +17,5 @@ def read_libraries():
@api_library.route("/<_id>", methods=["GET"])
def read_library(_id):
try:
library = LibraryController(_id)
except NoResultFound as e:
logger.debug(f"No result found for Library wid id {_id}")
logger.debug(f"Error {e}")
logger.debug(f"Error {dir(e)}")
return { "status": "error", "result": "Library not found"}, 404
return { "status": "ok", "result": library.data.to_dict() }, 200

View File

@@ -4,47 +4,24 @@ from sqlalchemy.exc import IntegrityError, NoResultFound
from .blueprint import api_library
from ....controller import LibraryController
from ....controller.functions import update_item_key
from ....schema.library.library import Library
import logging
logger = logging.getLogger(__name__)
def update_library_item(library:Library, key, value):
if key == "id":
raise AttributeError("id is not updatable")
try:
library.__getattribute__(key)
except AttributeError:
raise AttributeError(f"{key} not in library")
library.__setattr__(key, value)
@api_library.route("/<_id>", methods=["PATCH"])
def update_library(_id):
try:
data:dict = request.json
logger.debug(f"data: {data}")
except Exception as e:
logger.debug(f"{e}")
return { "status": "error", "error": "JSON Required" }, 415
try: # TODO: function
data:dict = request.json
controller = LibraryController(_id)
except NoResultFound as e:
logger.debug({e})
return { "status": "error", "error": "Library not found" }, 404
library = controller.data
for key, value in data.items():
try:
update_library_item(library, key, value)
except AttributeError as e:
logger.debug(f"Error updating {e}")
return { "status": "error", "error": e.name }
update_item_key(library, key, value)
try:
controller.update()
except IntegrityError as e:
logger.debug(f"DB Error Creating {e}")
return { "status": "error", "error": f"{e.orig}" }, 400
else:
return { "status": "ok", "result": controller.data.to_dict() }, 200

View File

@@ -3,5 +3,5 @@ query = sqlite:///test.db
[App]
port = 15012
debug = False
debug = True

BIN
test.db

Binary file not shown.

View File

@@ -7,11 +7,13 @@ from app.api.actions import install
from app.schema.library import Library, Tag, Book, BookTag, Path, Env
from app.controller import LibraryController
from app.controller.exceptions import LibraryReadException
import logging
logging.basicConfig(level=logging.DEBUG)
logger = logging.getLogger(__name__)
class TestDB(unittest.TestCase):
class TestController(unittest.TestCase):
def setUp(self):
os.environ["DEV_URIA_BIBLIOGAME_CONFIG_DB"] = "sqlite:///"
@@ -141,7 +143,7 @@ class TestDB(unittest.TestCase):
library = LibraryController(1, engine=self.engine)
library.delete()
self.assertRaises(NoResultFound, LibraryController, 1, engine=self.engine)
self.assertRaises(LibraryReadException, LibraryController, 1, engine=self.engine)
if __name__ == "__main__":
unittest.main()