evyos-center-server/wag-managment-api-service-version-2
2
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

init defaults completed

Browse Source
This commit is contained in:
berkay
2024-11-10 20:14:13 +03:00
parent aeda315119
commit c42a19c262
37 changed files with 582 additions and 308 deletions
Download Patch File Download Diff File Expand all files Collapse all files

69
databases/sql_models/core_mixin.py
View File

@@ -1,6 +1,5 @@
import datetime
from decimal import Decimal
from typing import Union
from sqlalchemy import (
TIMESTAMP,
@@ -24,8 +23,6 @@ from sqlalchemy_mixins.repr import ReprMixin
from sqlalchemy_mixins.smartquery import SmartQueryMixin
from api_library.date_time_actions.date_functions import DateTimeLocal, client_arrow
from api_objects import EmployeeTokenObject, OccupantTokenObject
from api_objects.auth.token_objects import Credentials
from databases.sql_models.sql_operations import FilterAttributes
from databases.sql_models.postgres_database import Base
@@ -80,7 +77,7 @@ class CrudMixin(Base, SmartQueryMixin, SessionMixin, FilterAttributes):
"created_by_id",
]
creds: Credentials = None # The credentials to use in the model.
creds = None # The credentials to use in the model.
client_arrow: DateTimeLocal = None # The arrow to use in the model.
valid_record_dict: dict = {"active": True, "deleted": False}
valid_record_args = lambda class_: [class_.active == True, class_.deleted == False]
@@ -94,7 +91,7 @@ class CrudMixin(Base, SmartQueryMixin, SessionMixin, FilterAttributes):
@classmethod
def set_user_define_properties(
cls, token: Union[EmployeeTokenObject, OccupantTokenObject]
cls, token
):
cls.creds = token.credentials
cls.client_arrow = DateTimeLocal(is_client=True, timezone=token.timezone)
@@ -149,7 +146,6 @@ class CrudMixin(Base, SmartQueryMixin, SessionMixin, FilterAttributes):
formatted_date = client_arrow.get(str(val)).format(
"DD-MM-YYYY HH:mm:ss"
)
print(key, "isinstance(value_type, datetime) | ", formatted_date)
return str(formatted_date) if val else None
elif isinstance(value_type, bool):
return bool(val) if val else None
@@ -175,7 +171,7 @@ class CrudMixin(Base, SmartQueryMixin, SessionMixin, FilterAttributes):
"""
check_kwargs = cls.extract_system_fields(kwargs)
cls.pre_query = cls.query.filter(cls.expiry_ends < system_arrow.now().date())
already_record = cls.filter_by_one(**check_kwargs, system=True).data
already_record = cls.filter_by_one(system=True, **check_kwargs).data
cls.pre_query = None
if already_record:
if already_record.deleted:
@@ -202,9 +198,10 @@ class CrudMixin(Base, SmartQueryMixin, SessionMixin, FilterAttributes):
created_record = cls()
for key, value in check_kwargs.items():
setattr(created_record, key, value)
if getattr(cls.creds, 'person_id', None) and getattr(cls.creds, 'person_name', None):
cls.created_by_id = cls.creds.person_id
cls.created_by = cls.creds.person_name
created_record.flush()
cls.created_by_id = cls.creds.person_id
cls.created_by = cls.creds.person_name
return created_record
def update(self, **kwargs):
@@ -223,11 +220,13 @@ class CrudMixin(Base, SmartQueryMixin, SessionMixin, FilterAttributes):
setattr(self, key, value)
if is_confirmed_argument:
self.confirmed_by_id = self.creds.person_id
self.confirmed_by = self.creds.person_name
if getattr(self.creds, 'person_id', None) and getattr(self.creds, 'person_name', None):
self.confirmed_by_id = self.creds.person_id
self.confirmed_by = self.creds.person_name
else:
self.updated_by_id = self.creds.person_id
self.updated_by = self.creds.person_name
if getattr(self.creds, 'person_id', None) and getattr(self.creds, 'person_name', None):
self.updated_by_id = self.creds.person_id
self.updated_by = self.creds.person_name
self.flush()
return self
@@ -252,8 +251,6 @@ class CrudMixin(Base, SmartQueryMixin, SessionMixin, FilterAttributes):
exclude.extend(
list(set(self.__exclude__fields__ or []).difference(exclude))
)
for i in self.__system_default_model__:
print("i", str(i)[-2:])
exclude.extend(
[
element
@@ -278,27 +275,27 @@ class CrudMixin(Base, SmartQueryMixin, SessionMixin, FilterAttributes):
if value_of_database is not None:
return_dict[key] = value_of_database
all_arguments = [
record
for record in self.__class__.__dict__
if "_" not in record[0] and "id" not in record[-2:]
]
for all_argument in all_arguments:
column = getattr(self.__class__, all_argument)
is_populate = isinstance(column, InstrumentedAttribute) and not hasattr(
column, "foreign_keys"
)
if is_populate and all_argument in include_joins or []:
populate_arg = getattr(self, all_argument, None)
if isinstance(populate_arg, list):
return_dict[all_argument] = [
arg.get_dict() if arg else [] for arg in populate_arg
]
elif getattr(populate_arg, "get_dict", None):
return_dict[all_argument] = (
populate_arg.get_dict() if populate_arg else []
)
# all_arguments = [
# record
# for record in self.__class__.__dict__
# if "_" not in record[0] and "id" not in record[-2:]
# ]
#
# for all_argument in all_arguments:
# column = getattr(self.__class__, all_argument)
# is_populate = isinstance(column, InstrumentedAttribute) and not hasattr(
# column, "foreign_keys"
# )
# if is_populate and all_argument in include_joins or []:
# populate_arg = getattr(self, all_argument, None)
# if isinstance(populate_arg, list):
# return_dict[all_argument] = [
# arg.get_dict() if arg else [] for arg in populate_arg
# ]
# elif getattr(populate_arg, "get_dict", None):
# return_dict[all_argument] = (
# populate_arg.get_dict() if populate_arg else []
# )
return dict(sorted(return_dict.items(), reverse=False))