Berkay
/
wag-managment-api-service-version-5
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

updated Query options

This commit is contained in:
berkay 2025-01-30 14:24:42 +03:00
parent 822e4155a1
commit b664f64eb4
35 changed files with 852 additions and 589 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

89
.idea/workspace.xml
View File

@ -6,29 +6,37 @@
<component name="ChangeListManager">
<list default="true" id="b5202e0c-6ddf-4a56-a13a-e18798c4c7cf" name="Changes" comment="">
<change beforePath="$PROJECT_DIR$/.idea/workspace.xml" beforeDir="false" afterPath="$PROJECT_DIR$/.idea/workspace.xml" afterDir="false" />
<change beforePath="$PROJECT_DIR$/ApiLayers/AllConfigs/main.py" beforeDir="false" afterPath="$PROJECT_DIR$/ApiLayers/AllConfigs/main.py" afterDir="false" />
<change beforePath="$PROJECT_DIR$/ApiLayers/AllConfigs/Redis/configs.py" beforeDir="false" afterPath="$PROJECT_DIR$/ApiLayers/AllConfigs/Redis/configs.py" afterDir="false" />
<change beforePath="$PROJECT_DIR$/ApiLayers/ApiServices/Token/token_handler.py" beforeDir="false" afterPath="$PROJECT_DIR$/ApiLayers/ApiServices/Token/token_handler.py" afterDir="false" />
<change beforePath="$PROJECT_DIR$/ApiLayers/ApiValidations/Custom/token_objects.py" beforeDir="false" afterPath="$PROJECT_DIR$/ApiLayers/ApiValidations/Custom/token_objects.py" afterDir="false" />
<change beforePath="$PROJECT_DIR$/ApiLayers/ApiValidations/Request/authentication.py" beforeDir="false" afterPath="$PROJECT_DIR$/ApiLayers/ApiValidations/Request/authentication.py" afterDir="false" />
<change beforePath="$PROJECT_DIR$/DockerApiServices/AuthServiceApi/Dockerfile" beforeDir="false" afterPath="$PROJECT_DIR$/DockerApiServices/AuthServiceApi/Dockerfile" afterDir="false" />
<change beforePath="$PROJECT_DIR$/ApiLayers/ApiValidations/Request/base_validations.py" beforeDir="false" afterPath="$PROJECT_DIR$/ApiLayers/ApiValidations/Request/base_validations.py" afterDir="false" />
<change beforePath="$PROJECT_DIR$/ApiLayers/ApiValidations/Response/default_response.py" beforeDir="false" afterPath="$PROJECT_DIR$/ApiLayers/ApiValidations/Response/default_response.py" afterDir="false" />
<change beforePath="$PROJECT_DIR$/ApiLayers/LanguageModels/Request/Auth/login.py" beforeDir="false" afterPath="$PROJECT_DIR$/ApiLayers/LanguageModels/Request/Auth/login.py" afterDir="false" />
<change beforePath="$PROJECT_DIR$/ApiLayers/LanguageModels/Response/authentication/auth.py" beforeDir="false" afterPath="$PROJECT_DIR$/ApiLayers/LanguageModels/Response/authentication/auth.py" afterDir="false" />
<change beforePath="$PROJECT_DIR$/ApiLayers/LanguageModels/set_defaults/language_setters.py" beforeDir="false" afterPath="$PROJECT_DIR$/ApiLayers/LanguageModels/set_defaults/language_setters.py" afterDir="false" />
<change beforePath="$PROJECT_DIR$/ApiLayers/LanguageModels/set_defaults/static_validation_retriever.py" beforeDir="false" afterPath="$PROJECT_DIR$/ApiLayers/LanguageModels/set_defaults/static_validation_retriever.py" afterDir="false" />
<change beforePath="$PROJECT_DIR$/ApiLayers/Schemas/identity/identity.py" beforeDir="false" afterPath="$PROJECT_DIR$/ApiLayers/Schemas/identity/identity.py" afterDir="false" />
<change beforePath="$PROJECT_DIR$/DockerApiServices/AuthServiceApi/config.py" beforeDir="false" afterPath="$PROJECT_DIR$/DockerApiServices/AuthServiceApi/config.py" afterDir="false" />
<change beforePath="$PROJECT_DIR$/DockerApiServices/AuthServiceApi/create_routes.py" beforeDir="false" afterPath="$PROJECT_DIR$/DockerApiServices/AuthServiceApi/create_routes.py" afterDir="false" />
<change beforePath="$PROJECT_DIR$/DockerApiServices/EventServiceApi/create_routes.py" beforeDir="false" afterPath="$PROJECT_DIR$/DockerApiServices/EventServiceApi/create_routes.py" afterDir="false" />
<change beforePath="$PROJECT_DIR$/DockerApiServices/InitServiceApi/config.py" beforeDir="false" afterPath="$PROJECT_DIR$/DockerApiServices/InitServiceApi/config.py" afterDir="false" />
<change beforePath="$PROJECT_DIR$/DockerApiServices/ValidationServiceApi/Dockerfile" beforeDir="false" afterPath="$PROJECT_DIR$/DockerApiServices/ValidationServiceApi/Dockerfile" afterDir="false" />
<change beforePath="$PROJECT_DIR$/DockerApiServices/ValidationServiceApi/config.py" beforeDir="false" afterPath="$PROJECT_DIR$/DockerApiServices/ValidationServiceApi/config.py" afterDir="false" />
<change beforePath="$PROJECT_DIR$/DockerApiServices/ValidationServiceApi/create_file.py" beforeDir="false" afterPath="$PROJECT_DIR$/DockerApiServices/ValidationServiceApi/create_file.py" afterDir="false" />
<change beforePath="$PROJECT_DIR$/DockerApiServices/ValidationServiceApi/create_routes.py" beforeDir="false" afterPath="$PROJECT_DIR$/DockerApiServices/ValidationServiceApi/create_routes.py" afterDir="false" />
<change beforePath="$PROJECT_DIR$/Events/AllEvents/authentication/auth/api_events.py" beforeDir="false" afterPath="$PROJECT_DIR$/Events/AllEvents/authentication/auth/api_events.py" afterDir="false" />
<change beforePath="$PROJECT_DIR$/Events/AllEvents/authentication/auth/auth.py" beforeDir="false" afterPath="$PROJECT_DIR$/Events/AllEvents/authentication/auth/auth.py" afterDir="false" />
<change beforePath="$PROJECT_DIR$/Events/AllEvents/authentication/auth/function_handlers.py" beforeDir="false" afterPath="$PROJECT_DIR$/Events/AllEvents/authentication/auth/function_handlers.py" afterDir="false" />
<change beforePath="$PROJECT_DIR$/Events/AllEvents/authentication/auth/info.py" beforeDir="false" afterPath="$PROJECT_DIR$/Events/AllEvents/authentication/auth/info.py" afterDir="false" />
<change beforePath="$PROJECT_DIR$/Events/AllEvents/authentication/auth/models.py" beforeDir="false" afterPath="$PROJECT_DIR$/Events/AllEvents/authentication/auth/models.py" afterDir="false" />
<change beforePath="$PROJECT_DIR$/Events/AllEvents/events/account/account_records.py" beforeDir="false" afterPath="$PROJECT_DIR$/Events/AllEvents/events/account/account_records.py" afterDir="false" />
<change beforePath="$PROJECT_DIR$/Events/AllEvents/template/template/api_events.py" beforeDir="false" afterPath="$PROJECT_DIR$/Events/AllEvents/template/template/api_events.py" afterDir="false" />
<change beforePath="$PROJECT_DIR$/Events/AllEvents/template/template/info.py" beforeDir="false" afterPath="$PROJECT_DIR$/Events/AllEvents/template/template/info.py" afterDir="false" />
<change beforePath="$PROJECT_DIR$/Events/AllEvents/template/template/models.py" beforeDir="false" afterPath="$PROJECT_DIR$/Events/AllEvents/template/template/models.py" afterDir="false" />
<change beforePath="$PROJECT_DIR$/Events/AllEvents/validations/__init__.py" beforeDir="false" afterPath="$PROJECT_DIR$/Events/AllEvents/validations/__init__.py" afterDir="false" />
<change beforePath="$PROJECT_DIR$/Events/AllEvents/validations/available/endpoints.py" beforeDir="false" />
<change beforePath="$PROJECT_DIR$/Events/AllEvents/validations/available/models.py" beforeDir="false" />
<change beforePath="$PROJECT_DIR$/Events/AllEvents/validations/validation/api_events.py" beforeDir="false" afterPath="$PROJECT_DIR$/Events/AllEvents/validations/validation/api_events.py" afterDir="false" />
<change beforePath="$PROJECT_DIR$/Events/AllEvents/validations/validation/cluster.py" beforeDir="false" afterPath="$PROJECT_DIR$/Events/AllEvents/validations/validation/cluster.py" afterDir="false" />
<change beforePath="$PROJECT_DIR$/Events/AllEvents/validations/validation/function_handlers.py" beforeDir="false" afterPath="$PROJECT_DIR$/Events/AllEvents/validations/validation/function_handlers.py" afterDir="false" />
<change beforePath="$PROJECT_DIR$/Events/AllEvents/validations/validation/info.py" beforeDir="false" afterPath="$PROJECT_DIR$/Events/AllEvents/validations/validation/info.py" afterDir="false" />
<change beforePath="$PROJECT_DIR$/Events/AllEvents/validations/validation/models.py" beforeDir="false" afterPath="$PROJECT_DIR$/Events/AllEvents/validations/validation/models.py" afterDir="false" />
<change beforePath="$PROJECT_DIR$/Events/AllEvents/validations/validation/validation.py" beforeDir="false" afterPath="$PROJECT_DIR$/Events/AllEvents/validations/validation/validation.py" afterDir="false" />
<change beforePath="$PROJECT_DIR$/Events/Engine/abstract_class.py" beforeDir="false" afterPath="$PROJECT_DIR$/Events/Engine/abstract_class.py" afterDir="false" />
<change beforePath="$PROJECT_DIR$/Events/Engine/set_defaults/setClusters.py" beforeDir="false" afterPath="$PROJECT_DIR$/Events/Engine/set_defaults/setClusters.py" afterDir="false" />
<change beforePath="$PROJECT_DIR$/docker-compose-services.yml" beforeDir="false" afterPath="$PROJECT_DIR$/docker-compose-services.yml" afterDir="false" />
<change beforePath="$PROJECT_DIR$/Services/PostgresDb/Models/filter_functions.py" beforeDir="false" afterPath="$PROJECT_DIR$/Services/PostgresDb/Models/filter_functions.py" afterDir="false" />
<change beforePath="$PROJECT_DIR$/Services/PostgresDb/Models/pagination.py" beforeDir="false" afterPath="$PROJECT_DIR$/Services/PostgresDb/Models/pagination.py" afterDir="false" />
<change beforePath="$PROJECT_DIR$/Services/PostgresDb/how_to.py" beforeDir="false" afterPath="$PROJECT_DIR$/Services/PostgresDb/how_to.py" afterDir="false" />
</list>
<option name="SHOW_DIALOG" value="false" />
<option name="HIGHLIGHT_CONFLICTS" value="true" />
@ -55,14 +63,17 @@
</component>
<component name="PropertiesComponent"><![CDATA[{
"keyToString": {
"Python.how_to.executor": "Run",
"Python.identity.executor": "Run",
"RunOnceActivity.ShowReadmeOnStart": "true",
"RunOnceActivity.git.unshallow": "true",
"git-widget-placeholder": "development",
"last_opened_file_path": "/home/berkay/git-gitea-evyos/wag-managment-api-service-version-5/Events/AllEvents/validations/validation"
"last_opened_file_path": "/home/berkay/git-gitea-evyos/wag-managment-api-service-version-5/Events/AllEvents/events/account"
}
}]]></component>
<component name="RecentsManager">
<key name="CopyFile.RECENT_KEYS">
<recent name="$PROJECT_DIR$/Events/AllEvents/events/account" />
<recent name="$PROJECT_DIR$/Events/AllEvents/validations/validation" />
<recent name="$PROJECT_DIR$/Events/AllEvents/template/template" />
<recent name="$PROJECT_DIR$/ApiLayers/LanguageModels/templates" />
@ -71,7 +82,7 @@
<recent name="$PROJECT_DIR$/ApiLayers/AllConfigs" />
</key>
</component>
<component name="RunManager">
<component name="RunManager" selected="Python.how_to">
<configuration name="events_file" type="PythonConfigurationType" factoryName="Python" temporary="true" nameIsGenerated="true">
<module name="wag-managment-api-service-version-5" />
<option name="ENV_FILES" value="" />
@ -94,8 +105,54 @@
<option name="INPUT_FILE" value="" />
<method v="2" />
</configuration>
<configuration name="how_to" type="PythonConfigurationType" factoryName="Python" temporary="true" nameIsGenerated="true">
<module name="wag-managment-api-service-version-5" />
<option name="ENV_FILES" value="" />
<option name="INTERPRETER_OPTIONS" value="" />
<option name="PARENT_ENVS" value="true" />
<envs>
<env name="PYTHONUNBUFFERED" value="1" />
</envs>
<option name="SDK_HOME" value="" />
<option name="WORKING_DIRECTORY" value="$PROJECT_DIR$/Services/PostgresDb" />
<option name="IS_MODULE_SDK" value="true" />
<option name="ADD_CONTENT_ROOTS" value="true" />
<option name="ADD_SOURCE_ROOTS" value="true" />
<option name="SCRIPT_NAME" value="$PROJECT_DIR$/Services/PostgresDb/how_to.py" />
<option name="PARAMETERS" value="" />
<option name="SHOW_COMMAND_LINE" value="false" />
<option name="EMULATE_TERMINAL" value="false" />
<option name="MODULE_MODE" value="false" />
<option name="REDIRECT_INPUT" value="false" />
<option name="INPUT_FILE" value="" />
<method v="2" />
</configuration>
<configuration name="identity" type="PythonConfigurationType" factoryName="Python" temporary="true" nameIsGenerated="true">
<module name="wag-managment-api-service-version-5" />
<option name="ENV_FILES" value="" />
<option name="INTERPRETER_OPTIONS" value="" />
<option name="PARENT_ENVS" value="true" />
<envs>
<env name="PYTHONUNBUFFERED" value="1" />
</envs>
<option name="SDK_HOME" value="" />
<option name="WORKING_DIRECTORY" value="$PROJECT_DIR$/ApiLayers/Schemas/identity" />
<option name="IS_MODULE_SDK" value="true" />
<option name="ADD_CONTENT_ROOTS" value="true" />
<option name="ADD_SOURCE_ROOTS" value="true" />
<option name="SCRIPT_NAME" value="$PROJECT_DIR$/ApiLayers/Schemas/identity/identity.py" />
<option name="PARAMETERS" value="" />
<option name="SHOW_COMMAND_LINE" value="false" />
<option name="EMULATE_TERMINAL" value="false" />
<option name="MODULE_MODE" value="false" />
<option name="REDIRECT_INPUT" value="false" />
<option name="INPUT_FILE" value="" />
<method v="2" />
</configuration>
<recent_temporary>
<list>
<item itemvalue="Python.how_to" />
<item itemvalue="Python.identity" />
<item itemvalue="Python.events_file" />
</list>
</recent_temporary>

28
ApiLayers/AllConfigs/Redis/configs.py
View File

@ -18,8 +18,6 @@ class WagRedis:
)
class RedisValidationKeys:
ENDPOINTS: str = "ENDPOINTS"
VALIDATIONS: str = "VALIDATIONS"
@ -51,17 +49,29 @@ class RedisCategoryKeys:
PAGE_MAPPER: str = "PAGE_MAPPER"
MENU_MAPPER: str = "MENU_MAPPER"
class RedisValidationKeysAction:
# LANGUAGE_MODELS:DYNAMIC:VALIDATIONS:
dynamic_validation_key: str = f"{RedisValidationKeys.LANGUAGE_MODELS}:{RedisValidationKeys.DYNAMIC}:{RedisValidationKeys.VALIDATIONS}"
dynamic_validation_key: str = (
f"{RedisValidationKeys.LANGUAGE_MODELS}:{RedisValidationKeys.DYNAMIC}:{RedisValidationKeys.VALIDATIONS}"
)
# LANGUAGE_MODELS:DYNAMIC:HEADERS:REQUEST
dynamic_header_request_key: str = f"{RedisValidationKeys.LANGUAGE_MODELS}:{RedisValidationKeys.DYNAMIC}:{RedisValidationKeys.HEADERS}:{RedisValidationKeys.REQUESTS}"
dynamic_header_request_key: str = (
f"{RedisValidationKeys.LANGUAGE_MODELS}:{RedisValidationKeys.DYNAMIC}:{RedisValidationKeys.HEADERS}:{RedisValidationKeys.REQUESTS}"
)
# LANGUAGE_MODELS:DYNAMIC:HEADERS:RESPONSE
dynamic_header_response_key: str = f"{RedisValidationKeys.LANGUAGE_MODELS}:{RedisValidationKeys.DYNAMIC}:{RedisValidationKeys.HEADERS}:{RedisValidationKeys.RESPONSES}"
dynamic_header_response_key: str = (
f"{RedisValidationKeys.LANGUAGE_MODELS}:{RedisValidationKeys.DYNAMIC}:{RedisValidationKeys.HEADERS}:{RedisValidationKeys.RESPONSES}"
)
# LANGUAGE_MODELS:STATIC:ERRORCODES:
static_error_code_key: str = f"{RedisValidationKeys.LANGUAGE_MODELS}:{RedisValidationKeys.STATIC}:{RedisValidationKeys.ERRORCODES}"
static_error_code_key: str = (
f"{RedisValidationKeys.LANGUAGE_MODELS}:{RedisValidationKeys.STATIC}:{RedisValidationKeys.ERRORCODES}"
)
# LANGUAGE_MODELS:STATIC:RESPONSES:
static_response_key: str = f"{RedisValidationKeys.LANGUAGE_MODELS}:{RedisValidationKeys.STATIC}:{RedisValidationKeys.RESPONSES}"
static_response_key: str = (
f"{RedisValidationKeys.LANGUAGE_MODELS}:{RedisValidationKeys.STATIC}:{RedisValidationKeys.RESPONSES}"
)
# LANGUAGE_MODELS:STATIC:REQUESTS:
static_request_key: str = f"{RedisValidationKeys.LANGUAGE_MODELS}:{RedisValidationKeys.STATIC}:{RedisValidationKeys.REQUESTS}"
static_request_key: str = (
f"{RedisValidationKeys.LANGUAGE_MODELS}:{RedisValidationKeys.STATIC}:{RedisValidationKeys.REQUESTS}"
)

177
ApiLayers/ApiServices/Token/token_handler.py
View File

@ -9,7 +9,7 @@ from ApiLayers.ApiLibrary.common.line_number import get_line_number_for_error
from ApiLayers.ApiLibrary.date_time_actions.date_functions import DateTimeLocal
from ApiLayers.ApiLibrary.token.password_module import PasswordModule
from ApiLayers.ErrorHandlers import HTTPExceptionApi
from ApiLayers.Schemas.identity.identity import UsersTokens, People
from ApiLayers.ApiValidations.Custom.token_objects import (
EmployeeTokenObject,
OccupantTokenObject,
@ -37,6 +37,7 @@ from Services.Redis import RedisActions, AccessToken
if TYPE_CHECKING:
from fastapi import Request
T = TypeVar("T", EmployeeTokenObject, OccupantTokenObject)
@ -55,6 +56,85 @@ class TokenService:
"""Get all tokens for a user from Redis."""
return RedisActions.get_json(list_keys=[f"*:{str(user.uu_id)}"])
@classmethod
def do_employee_login(
cls, request: "Request", user: Users, domain: str
) -> Dict[str, Any]:
"""Handle employee login process and return login information."""
from ApiLayers.Schemas.identity.identity import UsersTokens, People
db_session = Employees.new_session()
list_employee = Employees.filter_all(
Employees.people_id == user.person_id, db=db_session
).data
companies_uu_id_list: List[str] = []
companies_id_list: List[int] = []
companies_list: List[Dict[str, Any]] = []
duty_uu_id_list: List[str] = []
duty_id_list: List[int] = []
for employee in list_employee:
staff = Staff.filter_one(Staff.id == employee.staff_id, db=db_session).data
if duties := Duties.filter_one(
Duties.id == staff.duties_id, db=db_session
).data:
if duty_found := Duty.filter_by_one(
id=duties.duties_id, db=db_session
).data:
duty_uu_id_list.append(str(duty_found.uu_id))
duty_id_list.append(duty_found.id)
department = Departments.filter_one(
Departments.id == duties.department_id, db=db_session
).data
if company := Companies.filter_one(
Companies.id == department.company_id, db=db_session
).data:
companies_uu_id_list.append(str(company.uu_id))
companies_id_list.append(company.id)
company_address = Addresses.filter_by_one(
id=company.official_address_id, db=db_session
).data
companies_list.append(
{
"uu_id": str(company.uu_id),
"public_name": company.public_name,
"company_type": company.company_type,
"company_address": company_address,
}
)
person = People.filter_one(People.id == user.person_id, db=db_session).data
model_value = EmployeeTokenObject(
domain=domain,
user_type=UserType.employee.value,
user_uu_id=str(user.uu_id),
credentials=user.credentials(),
user_id=user.id,
person_id=person.id,
person_uu_id=str(person.uu_id),
full_name=person.full_name,
request=dict(request.headers),
companies_uu_id_list=companies_uu_id_list,
companies_id_list=companies_id_list,
duty_uu_id_list=duty_uu_id_list,
duty_id_list=duty_id_list,
timezone=user.local_timezone or "GMT+0",
lang="tr",
).model_dump()
if access_token := cls.set_object_to_redis(user, model_value):
return {
"access_token": access_token,
"user_type": UserType.employee.name,
"selection_list": companies_list,
}
raise HTTPExceptionApi(
error_code="",
lang="en",
loc=get_line_number_for_error(),
sys_msg="Creating Token failed...",
)
@classmethod
def do_occupant_login(
cls, request: "Request", user: Users, domain: str
@ -113,14 +193,17 @@ class TokenService:
}
else:
occupants_selection_dict[build_key]["occupants"].append(occupant_data)
person = user.person
model_value = OccupantTokenObject(
domain=domain,
user_type=UserType.occupant.value,
user_uu_id=str(user.uu_id),
credentials=user.credentials(),
user_id=user.id,
person_id=user.person_id,
person_uu_id=str(user.person.uu_id),
person_id=person.id,
person_uu_id=str(person.uu_id),
full_name=person.full_name,
request=dict(request.headers),
available_occupants=occupants_selection_dict,
timezone=user.local_timezone or "GMT+0",
@ -180,84 +263,6 @@ class TokenService:
sys_msg="Saving Token failed...",
)
@classmethod
def do_employee_login(
cls, request: "Request", user: Users, domain: str
) -> Dict[str, Any]:
"""Handle employee login process and return login information."""
db_session = Employees.new_session()
list_employee = Employees.filter_all(
Employees.people_id == user.person_id, db=db_session
).data
companies_uu_id_list: List[str] = []
companies_id_list: List[int] = []
companies_list: List[Dict[str, Any]] = []
duty_uu_id_list: List[str] = []
duty_id_list: List[int] = []
for employee in list_employee:
staff = Staff.filter_one(Staff.id == employee.staff_id, db=db_session).data
if duties := Duties.filter_one(
Duties.id == staff.duties_id, db=db_session
).data:
if duty_found := Duty.filter_by_one(
id=duties.duties_id, db=db_session
).data:
duty_uu_id_list.append(str(duty_found.uu_id))
duty_id_list.append(duty_found.id)
department = Departments.filter_one(
Departments.id == duties.department_id, db=db_session
).data
if company := Companies.filter_one(
Companies.id == department.company_id, db=db_session
).data:
companies_uu_id_list.append(str(company.uu_id))
companies_id_list.append(company.id)
company_address = Addresses.filter_by_one(
id=company.official_address_id, db=db_session
).data
companies_list.append(
{
"uu_id": str(company.uu_id),
"public_name": company.public_name,
"company_type": company.company_type,
"company_address": company_address,
}
)
person = People.filter_one(People.id == user.person_id, db=db_session).data
model_value = EmployeeTokenObject(
domain=domain,
user_type=UserType.employee.value,
user_uu_id=str(user.uu_id),
credentials=user.credentials(),
user_id=user.id,
person_id=person.id,
person_uu_id=str(person.uu_id),
request=dict(request.headers),
companies_uu_id_list=companies_uu_id_list,
companies_id_list=companies_id_list,
duty_uu_id_list=duty_uu_id_list,
duty_id_list=duty_id_list,
timezone=user.local_timezone or "GMT+0",
lang="tr",
).model_dump()
if access_token := cls.set_object_to_redis(user, model_value):
return {
"access_token": access_token,
"user_type": UserType.employee.name,
"selection_list": companies_list,
}
raise HTTPExceptionApi(
error_code="",
lang="en",
loc=get_line_number_for_error(),
sys_msg="Creating Token failed...",
)
@classmethod
def remove_token_with_domain(cls, user: Users, domain: str) -> None:
"""Remove all tokens for a user with specific domain."""
@ -282,6 +287,8 @@ class TokenService:
) -> Dict[str, Any]:
"""Set access token to redis and handle user session."""
from ApiLayers.AllConfigs.Token.config import Auth
from ApiLayers.Schemas.identity.identity import UsersTokens, People
cls.remove_token_with_domain(user=user, domain=domain)
# Users.client_arrow = DateTimeLocal(is_client=True, timezone=user.local_timezone)
login_dict, db_session = {}, UsersTokens.new_session()
@ -309,9 +316,13 @@ class TokenService:
users_token.token = users_token_created
users_token.save(db=db_session)
else:
if arrow.now() > arrow.get(str(users_token.expires_at)): # Check if token is expired
if arrow.now() > arrow.get(
str(users_token.expires_at)
): # Check if token is expired
users_token.token = users_token_created
users_token.expires_at = str(arrow.now().datetime + Auth.TOKEN_EXPIRE_DAY_1)
users_token.expires_at = str(
arrow.now().datetime + Auth.TOKEN_EXPIRE_DAY_1
)
users_token.save(db=db_session)
else:
login_dict["refresh_token"] = users_token.token

1
ApiLayers/ApiValidations/Custom/token_objects.py
View File

@ -33,6 +33,7 @@ class ApplicationToken(BaseModel):
person_id: int
person_uu_id: str
full_name: Optional[str] = None
request: Optional[dict] = None # Request Info of Client
expires_at: Optional[float] = None # Expiry timestamp

4
ApiLayers/ApiValidations/Request/authentication.py
View File

@ -51,7 +51,9 @@ class CreatePassword(BaseModelRegular, CreatePasswordValidation):
class OccupantSelection(BaseModel):
build_living_space_uu_id: str = Field(..., example="987fcdeb-51a2-43e7-9876-543210987654")
build_living_space_uu_id: str = Field(
..., example="987fcdeb-51a2-43e7-9876-543210987654"
)
model_config = ConfigDict(
json_schema_extra={

5
ApiLayers/ApiValidations/Request/base_validations.py
View File

@ -5,9 +5,8 @@ from ApiLayers.ApiValidations.handler import BaseModelRegular
class ListOptions(BaseModelRegular):
page: Optional[int] = 1
size: Optional[int] = 10
order_field: Optional[str] = "id"
order_type: Optional[str] = "asc"
include_joins: Optional[list] = None
order_field: Optional[list[str]] = None
order_type: Optional[list[str]] = None
query: Optional[dict] = None

4
ApiLayers/ApiValidations/Response/default_response.py
View File

@ -2,7 +2,9 @@ from typing import Optional
from fastapi import status
from fastapi.responses import JSONResponse
from ApiLayers.LanguageModels.set_defaults.static_validation_retriever import StaticValidationRetriever
from ApiLayers.LanguageModels.set_defaults.static_validation_retriever import (
StaticValidationRetriever,
)
class BaseEndpointResponse(StaticValidationRetriever):

4
ApiLayers/LanguageModels/Response/authentication/auth.py
View File

@ -1,5 +1,3 @@
authResponses = {
"LOGIN_SELECT": {
"tr": {
@ -96,5 +94,5 @@ authResponses = {
"en": {
"message": "Token successfully refreshed.",
},
}
},
}

49
ApiLayers/LanguageModels/set_defaults/language_setters.py
View File

@ -1,4 +1,7 @@
from ApiLayers.AllConfigs.Redis.configs import RedisValidationKeysAction, RedisValidationKeys
from ApiLayers.AllConfigs.Redis.configs import (
RedisValidationKeysAction,
RedisValidationKeys,
)
from ApiLayers.AllConfigs.main import LanguageConfig
from Events.Engine.set_defaults.category_cluster_models import CategoryClusterController
from Services.Redis.Actions.actions import RedisActions
@ -28,7 +31,9 @@ class SetDefaultLanguageModelsRedis:
for lang in list(LanguageConfig.SUPPORTED_LANGUAGES):
for code, dict_to_set in response.items():
# [SAVE]REDIS => LANGUAGE_MODELS:STATIC:RESPONSES:{ResponseCode}:tr = {...}
set_key = f"{RedisValidationKeysAction.static_response_key}:{code}:{lang}"
set_key = (
f"{RedisValidationKeysAction.static_response_key}:{code}:{lang}"
)
RedisActions.set_json(list_keys=[set_key], value=dict_to_set[lang])
self.std_out += f"Language Response Models are set to Redis\n"
@ -81,10 +86,18 @@ class SetClusterLanguageModelsRedis:
self.std_out += f"Setting models from cluster : {cluster_control.name}\n"
for endpoint in cluster_control.category_cluster.ENDPOINTS.values():
for key_event, event in endpoint.EVENTS.items():
merged_language_dict = self.merge_language_dicts(event.LANGUAGE_MODELS)
request_validation = getattr(event.REQUEST_VALIDATOR, "model_fields", None)
response_validation = getattr(event.RESPONSE_VALIDATOR, "model_fields", None)
objects_missing = bool(request_validation) and bool(merged_language_dict)
merged_language_dict = self.merge_language_dicts(
event.LANGUAGE_MODELS
)
request_validation = getattr(
event.REQUEST_VALIDATOR, "model_fields", None
)
response_validation = getattr(
event.RESPONSE_VALIDATOR, "model_fields", None
)
objects_missing = bool(request_validation) and bool(
merged_language_dict
)
if not objects_missing:
continue
if merged_language_dict:
@ -105,17 +118,23 @@ class SetClusterLanguageModelsRedis:
[SAVE]REDIS => LANGUAGE_MODELS:DYNAMIC:HEADERS:REQUEST:{FunctionCode}:tr = {...}
Get Request BaseModel pydantic model_fields of each event and set headers which are included in model_fields
"""
for lang in list(LanguageConfig.SUPPORTED_LANGUAGES): # Iterate(languages ["tr", "en"])
for lang in list(
LanguageConfig.SUPPORTED_LANGUAGES
): # Iterate(languages ["tr", "en"])
for key_field in self.events_rq_dict.keys(): # Iterate(function_code)
request_model = self.events_rq_dict[key_field]
if not request_model:
self.std_out += f"Request validation model not found for {key_field}\n"
self.std_out += (
f"Request validation model not found for {key_field}\n"
)
continue
if (
key_field not in self.events_rq_dict
or key_field not in self.events_lm_dict
):
self.std_out += f"Request language model are missing {key_field}\n"
self.std_out += (
f"Request language model are missing {key_field}\n"
)
continue
value_to_set = {}
@ -130,17 +149,23 @@ class SetClusterLanguageModelsRedis:
[SAVE]REDIS => LANGUAGE_MODELS:DYNAMIC:HEADERS:RESPONSE:{FunctionCode}:en = {...}
Get Response BaseModel pydantic model_fields of each event and set headers which are included in model_fields
"""
for lang in list(LanguageConfig.SUPPORTED_LANGUAGES): # Iterate(languages ["tr", "en"])
for lang in list(
LanguageConfig.SUPPORTED_LANGUAGES
): # Iterate(languages ["tr", "en"])
for key_field in self.events_rs_dict.keys(): # Iterate(function_code)
response_model = self.events_rs_dict[key_field]
if not response_model:
self.std_out += f"Response validation model not found for {key_field}\n"
self.std_out += (
f"Response validation model not found for {key_field}\n"
)
continue
if (
key_field not in self.events_rs_dict
or key_field not in self.events_lm_dict
):
self.std_out += f"Response language model are missing {key_field}\n"
self.std_out += (
f"Response language model are missing {key_field}\n"
)
continue
value_to_set = {}

7
ApiLayers/LanguageModels/set_defaults/static_validation_retriever.py
View File

@ -15,9 +15,12 @@ class StaticValidationRetriever:
@property
def response(self) -> Optional[dict]:
language_model = RedisActions.get_json(
list_keys=[RedisValidationKeysAction.static_response_key, self.code, self.lang]
list_keys=[
RedisValidationKeysAction.static_response_key,
self.code,
self.lang,
]
)
if language_model.status:
return language_model.first
return {"message": f"{self.code} -> Language model not found"}

6
ApiLayers/Schemas/identity/identity.py
View File

@ -16,7 +16,7 @@ from sqlalchemy import (
from sqlalchemy.orm import mapped_column, relationship, Mapped
from Services.PostgresDb import CrudCollection
from config import ApiStatic
# from config import ApiStatic
from ApiLayers.ApiLibrary.date_time_actions.date_functions import system_arrow
from ApiLayers.ApiLibrary.extensions.select import (
@ -24,7 +24,7 @@ from ApiLayers.ApiLibrary.extensions.select import (
SelectActionWithEmployee,
)
from ApiLayers.AllConfigs.Token.config import Auth
from ApiLayers.ApiServices.Login.user_login_handler import UserLoginModule
from ApiLayers.ApiValidations.Request import InsertUsers, InsertPerson
from ApiLayers.LanguageModels.Database.identity.identity import (
UsersTokensLanguageModel,
@ -66,7 +66,7 @@ class UsersTokens(CrudCollection):
# users = relationship("Users", back_populates="tokens", foreign_keys=[user_id])
class Users(CrudCollection, UserLoginModule, SelectAction):
class Users(CrudCollection, SelectAction):
"""
Application User frame to connect to API with assigned token-based HTTP connection
"""

1
DockerApiServices/AuthServiceApi/config.py
View File

@ -69,4 +69,3 @@ class LanguageConfig:
SUPPORTED_LANGUAGES = ["en", "tr"]
DEFAULT_LANGUAGE = "tr"

1
DockerApiServices/InitServiceApi/config.py
View File

@ -70,6 +70,7 @@ class LanguageConfig:
SUPPORTED_LANGUAGES = ["en", "tr"]
DEFAULT_LANGUAGE = "tr"
class ValidationsConfig:
SUPPORTED_VALIDATIONS = ["header", "validation", "all"]

4
Events/AllEvents/authentication/auth/auth.py
View File

@ -169,7 +169,9 @@ AuthenticationChangePasswordEventMethods = MethodToEvent(
)
def authentication_change_password_event_callable(request: Request, data: EndpointBaseRequestModel):
def authentication_change_password_event_callable(
request: Request, data: EndpointBaseRequestModel
):
context_retriever = ContextRetrievers(
func=authentication_change_password_event_callable
)

93
Events/AllEvents/authentication/auth/function_handlers.py
View File

@ -288,25 +288,30 @@ class AuthenticationFunctions(BaseRouteModel):
"""Refresh user info using access token"""
if cls.context_retriever.token:
db = Users.new_session()
if found_user := Users.filter_one(Users.id == cls.context_retriever.token.user_id, db=db).data:
if found_user := Users.filter_one(
Users.id == cls.context_retriever.token.user_id, db=db
).data:
return EndpointSuccessResponse(
code="USER_INFO_REFRESHED", lang=cls.context_retriever.token.lang
).as_dict({
"access_token": cls.context_retriever.get_token, "user": found_user.get_dict(),
})
).as_dict(
{
"access_token": cls.context_retriever.get_token,
"user": found_user.get_dict(),
}
)
if not found_user:
return EndpointNotAcceptableResponse(
code="USER_NOT_FOUND", lang=cls.context_retriever.token.lang
).as_dict(
data={}
)
).as_dict(data={})
@classmethod # Requires no auth context
def authentication_change_password(cls, data: Any):
"""Change password with access token"""
if cls.context_retriever.token:
db = Users.new_session()
if found_user := Users.filter_one(Users.id == cls.context_retriever.token.user_id, db=db).data:
if found_user := Users.filter_one(
Users.id == cls.context_retriever.token.user_id, db=db
).data:
found_user.set_password(data.new_password)
return EndpointSuccessResponse(
code="PASSWORD_CHANGED", lang=cls.context_retriever.token.lang
@ -314,9 +319,7 @@ class AuthenticationFunctions(BaseRouteModel):
if not found_user:
return EndpointNotAcceptableResponse(
code="USER_NOT_FOUND", lang=cls.context_retriever.token.lang
).as_dict(
data={}
)
).as_dict(data={})
@classmethod # Requires not auth context
def authentication_create_password(cls, data: Any):
@ -325,11 +328,9 @@ class AuthenticationFunctions(BaseRouteModel):
if not data.re_password == data.password:
return EndpointNotAcceptableResponse(
code="PASSWORD_NOT_MATCH", lang=cls.context_retriever.token.lang
).as_dict(
data={"password": data.password, "re_password": data.re_password}
)
).as_dict(data={"password": data.password, "re_password": data.re_password})
if found_user := Users.filter_one(
Users.password_token == data.password_token, db=db
Users.password_token == data.password_token, db=db
).data:
found_user.create_password(found_user=found_user, password=data.password)
found_user.password_token = ""
@ -346,11 +347,9 @@ class AuthenticationFunctions(BaseRouteModel):
Users.id == cls.context_retriever.token.user_id, db=db
).data
if not found_user:
return EndpointNotAcceptableResponse(
return EndpointNotAcceptableResponse(
code="USER_NOT_FOUND", lang=cls.context_retriever.token.lang
).as_dict(
data={}
)
).as_dict(data={})
registered_tokens = UsersTokens.filter_all(
UsersTokens.user_id == cls.context_retriever.token.user_id, db=db
)
@ -375,9 +374,7 @@ class AuthenticationFunctions(BaseRouteModel):
if not found_user:
return EndpointNotAcceptableResponse(
code="USER_NOT_FOUND", lang=cls.context_retriever.token.lang
).as_dict(
data={}
)
).as_dict(data={})
registered_tokens = UsersTokens.filter_all_system(
UsersTokens.user_id == cls.context_retriever.token.user_id,
UsersTokens.domain == cls.context_retriever.token.domain,
@ -404,6 +401,7 @@ class AuthenticationFunctions(BaseRouteModel):
"""
import arrow
from ApiLayers.ApiServices.Token.token_handler import TokenService
db = UsersTokens.new_session()
token_refresher: UsersTokens = UsersTokens.filter_by_one(
token=data.refresh_token,
@ -414,11 +412,11 @@ class AuthenticationFunctions(BaseRouteModel):
if not token_refresher:
return EndpointNotAcceptableResponse(
code="REFRESHER_NOT_FOUND", lang=language
).as_dict(
data={"refresh_token": data.refresh_token}
)
).as_dict(data={"refresh_token": data.refresh_token})
if found_user := Users.filter_one(Users.id == token_refresher.user_id, db=db).data:
if found_user := Users.filter_one(
Users.id == token_refresher.user_id, db=db
).data:
token_created = TokenService.set_access_token_to_redis(
request=request,
user=found_user,
@ -427,51 +425,63 @@ class AuthenticationFunctions(BaseRouteModel):
)
found_user.last_agent = request.headers.get("User-Agent", None)
found_user.last_platform = request.headers.get("Origin", None)
found_user.last_remote_addr = getattr(request, "remote_addr", None) or request.headers.get("X-Forwarded-For", None)
found_user.last_remote_addr = getattr(
request, "remote_addr", None
) or request.headers.get("X-Forwarded-For", None)
found_user.last_seen = str(arrow.now())
response_data = {
"access_token": token_created.get("access_token"),
"refresh_token": data.refresh_token,
}
return EndpointSuccessResponse(code="TOKEN_REFRESH", lang=language).as_dict(data=response_data)
return EndpointSuccessResponse(code="TOKEN_REFRESH", lang=language).as_dict(
data=response_data
)
raise EndpointNotAcceptableResponse(
code="USER_NOT_FOUND", lang=language
).as_dict(
data={}
)
).as_dict(data={})
@classmethod # Requires not auth context
def authentication_forgot_password(cls, data: Any):
"""Send an email to user for a valid password reset token"""
import arrow
from ApiLayers.ApiServices.Token.token_handler import TokenService
from ApiLayers.AllConfigs.Templates.password_templates import change_your_password_template
from ApiLayers.AllConfigs.Templates.password_templates import (
change_your_password_template,
)
from Services.Email.send_email import email_sender
from config import ApiStatic
db = Users.new_session()
request = cls.context_retriever.request
found_user: Users = Users.check_user_exits(access_key=data.access_key, domain=data.domain)
found_user: Users = Users.check_user_exits(
access_key=data.access_key, domain=data.domain
)
forgot_key = TokenService._create_access_token(access=False)
forgot_link = ApiStatic.forgot_link(forgot_key=forgot_key)
send_email_completed = email_sender.send_email(
subject=f"Dear {found_user.user_tag}, your forgot password link has been sent.",
receivers=[str(found_user.email)],
html=change_your_password_template(user_name=found_user.user_tag, forgot_link=forgot_link),
html=change_your_password_template(
user_name=found_user.user_tag, forgot_link=forgot_link
),
)
if not send_email_completed:
return EndpointBadRequestResponse(
code="EMAIL_NOT_SENT", lang=cls.context_retriever.token.lang
).as_dict(
data={"email": found_user.email}
)
).as_dict(data={"email": found_user.email})
found_user.password_token = forgot_key
found_user.password_token_is_valid = str(arrow.now().shift(days=1))
found_user.save(db=db)
return EndpointSuccessResponse(
code="FORGOT_PASSWORD", lang=cls.context_retriever.token.lang
).as_dict(data={"user": found_user.get_dict(), "forgot_link": forgot_link, "token": forgot_key})
).as_dict(
data={
"user": found_user.get_dict(),
"forgot_link": forgot_link,
"token": forgot_key,
}
)
@classmethod # Requires not auth context
def authentication_reset_password(cls, data: Any):
@ -482,12 +492,15 @@ class AuthenticationFunctions(BaseRouteModel):
def authentication_download_avatar(cls):
"""Download avatar icon and profile info of user"""
import arrow
db = Users.new_session()
if found_user := Users.filter_one(
Users.id == cls.context_retriever.token.user_id, db=db
Users.id == cls.context_retriever.token.user_id, db=db
).data:
expired_starts = str(arrow.now() - arrow.get(str(found_user.expiry_ends)))
expired_int = arrow.now().datetime - arrow.get(str(found_user.expiry_ends)).datetime
expired_int = (
arrow.now().datetime - arrow.get(str(found_user.expiry_ends)).datetime
)
user_info = {
"lang": cls.context_retriever.token.lang,
"full_name": found_user.person.full_name,

2
Events/AllEvents/authentication/auth/info.py
View File

@ -7,5 +7,5 @@ authentication_page_info = PageInfo(
description={"en": "Authentication"},
icon="",
parent="",
url=""
url="",
)

3
Events/AllEvents/authentication/auth/models.py
View File

@ -13,7 +13,8 @@ from ApiLayers.ApiValidations.Request import (
class AuthenticationRequestModels:
LoginSuperUserRequestModel = Login
SelectCompanyOrOccupantTypeSuperUserRequestModel = {
"EmployeeSelection": EmployeeSelection, "OccupantSelection": OccupantSelection,
"EmployeeSelection": EmployeeSelection,
"OccupantSelection": OccupantSelection,
}
RefresherRequestModel = Remember
LogoutRequestModel = Logout

372
Events/AllEvents/events/account/account_records.py
View File

@ -1,351 +1,39 @@
"""
Account records service implementation.
template related API endpoints.
"""
from typing import Union
from pydantic import Field
from typing import Any, Dict
from fastapi import Request
from ApiEvents.abstract_class import MethodToEvent, endpoint_wrapper
from ApiEvents.base_request_model import DictRequestModel
from ApiValidations.Custom.token_objects import (
OccupantTokenObject,
EmployeeTokenObject,
)
from ApiLibrary import system_arrow
from ApiValidations.Request.account_records import (
InsertAccountRecord,
UpdateAccountRecord,
)
from ApiValidations.Request.base_validations import ListOptions
from Schemas import (
BuildLivingSpace,
AccountRecords,
BuildIbans,
BuildDecisionBookPayments,
ApiEnumDropdown,
)
from Services.PostgresDb.Models.alchemy_response import (
AlchemyJsonResponse,
)
from ApiValidations.Response import AccountRecordResponse
from .models import (
InsertAccountRecordRequestModel,
UpdateAccountRecordRequestModel,
ListOptionsRequestModel,
from Events.Engine.abstract_class import MethodToEvent
from Events.base_request_model import EndpointBaseRequestModel, ContextRetrievers
from .account_records import template_event
AuthenticationLoginEventMethods = MethodToEvent(
name="AuthenticationLoginEventMethods",
events={
template_event.key: template_event,
},
headers=[],
errors=[],
url="/login",
method="POST",
summary="Login via domain and access key : [email] | [phone]",
description="Login to the system via domain, access key : [email] | [phone]",
)
class AccountListEventMethod(MethodToEvent):
event_type = "SELECT"
event_description = ""
event_category = ""
__event_keys__ = {
"7192c2aa-5352-4e36-98b3-dafb7d036a3d": "account_records_list",
"208e6273-17ef-44f0-814a-8098f816b63a": "account_records_list_flt_res",
}
__event_validation__ = {
"7192c2aa-5352-4e36-98b3-dafb7d036a3d": (
AccountRecordResponse,
[AccountRecords.__language_model__],
),
"208e6273-17ef-44f0-814a-8098f816b63a": (
AccountRecordResponse,
[AccountRecords.__language_model__],
),
}
@classmethod
def account_records_list(
cls,
list_options: ListOptionsRequestModel,
token_dict: Union[EmployeeTokenObject, OccupantTokenObject],
):
db_session = AccountRecords.new_session()
if isinstance(token_dict, OccupantTokenObject):
AccountRecords.pre_query = AccountRecords.filter_all(
AccountRecords.company_id
== token_dict.selected_occupant.responsible_company_id,
db=db_session,
).query
elif isinstance(token_dict, EmployeeTokenObject):
AccountRecords.pre_query = AccountRecords.filter_all(
AccountRecords.company_id == token_dict.selected_company.company_id,
db=db_session,
).query
AccountRecords.filter_attr = list_options
records = AccountRecords.filter_all(db=db_session)
return AlchemyJsonResponse(
completed=True,
message="Account records listed successfully",
result=records,
cls_object=AccountRecords,
filter_attributes=list_options,
response_model=AccountRecordResponse,
)
@classmethod
def account_records_list_flt_res(
cls,
list_options: ListOptionsRequestModel,
token_dict: Union[EmployeeTokenObject, OccupantTokenObject],
):
db_session = AccountRecords.new_session()
if not isinstance(token_dict, OccupantTokenObject):
raise AccountRecords.raise_http_exception(
status_code="HTTP_404_NOT_FOUND",
error_case="UNAUTHORIZED",
message="Only Occupant can see this data",
data={},
)
return_list = []
living_space: BuildLivingSpace = BuildLivingSpace.filter_by_one(
id=token_dict.selected_occupant.living_space_id
).data
if not living_space:
raise AccountRecords.raise_http_exception(
status_code="HTTP_404_NOT_FOUND",
error_case="UNAUTHORIZED",
message="Living space not found",
data={},
)
if not list_options:
list_options = ListOptions()
main_filters = [
AccountRecords.living_space_id
== token_dict.selected_occupant.living_space_id,
BuildDecisionBookPayments.process_date
>= str(system_arrow.now().shift(months=-3).date()),
BuildDecisionBookPayments.process_date
< str(system_arrow.find_last_day_of_month(living_space.expiry_ends)),
BuildDecisionBookPayments.process_date
>= str(system_arrow.get(living_space.expiry_starts)),
BuildDecisionBookPayments.is_confirmed == True,
AccountRecords.active == True,
]
order_type = "desc"
if list_options.order_type:
order_type = "asc" if list_options.order_type[0] == "a" else "desc"
order_by_list = BuildDecisionBookPayments.process_date.desc()
if list_options.order_field:
if list_options.order_field == "process_date":
order_by_list = (
BuildDecisionBookPayments.process_date.asc()
if order_type == "asc"
else BuildDecisionBookPayments.process_date.desc()
)
if list_options.order_field == "bank_date":
order_by_list = (
AccountRecords.bank_date.desc()
if order_type == "asc"
else AccountRecords.bank_date.asc()
)
if list_options.order_field == "currency_value":
order_by_list = (
AccountRecords.currency_value.desc()
if order_type == "asc"
else AccountRecords.currency_value.asc()
)
if list_options.order_field == "process_comment":
order_by_list = (
AccountRecords.process_comment.desc()
if order_type == "asc"
else AccountRecords.process_comment.asc()
)
if list_options.order_field == "payment_amount":
order_by_list = (
BuildDecisionBookPayments.payment_amount.desc()
if order_type == "asc"
else BuildDecisionBookPayments.payment_amount.asc()
)
if list_options.query:
for key, value in list_options.query.items():
if key == "process_date":
main_filters.append(BuildDecisionBookPayments.process_date == value)
if key == "bank_date":
main_filters.append(AccountRecords.bank_date == value)
if key == "currency":
main_filters.append(BuildDecisionBookPayments.currency == value)
if key == "currency_value":
main_filters.append(AccountRecords.currency_value == value)
if key == "process_comment":
main_filters.append(AccountRecords.process_comment == value)
if key == "payment_amount":
main_filters.append(
BuildDecisionBookPayments.payment_amount == value
)
query = (
AccountRecords.session.query(
BuildDecisionBookPayments.process_date,
BuildDecisionBookPayments.payment_amount,
BuildDecisionBookPayments.currency,
AccountRecords.bank_date,
AccountRecords.currency_value,
AccountRecords.process_comment,
BuildDecisionBookPayments.uu_id,
)
.join(
AccountRecords,
AccountRecords.id == BuildDecisionBookPayments.account_records_id,
)
.filter(*main_filters)
).order_by(order_by_list)
query.limit(list_options.size or 5).offset(
(list_options.page or 1 - 1) * list_options.size or 5
)
for list_of_values in query.all() or []:
return_list.append(
{
"process_date": list_of_values[0],
"payment_amount": list_of_values[1],
"currency": list_of_values[2],
"bank_date": list_of_values[3],
"currency_value": list_of_values[4],
"process_comment": list_of_values[5],
}
)
return AlchemyJsonResponse(
completed=True,
message="Account records listed successfully",
result=return_list,
cls_object=AccountRecords,
filter_attributes=list_options,
response_model=AccountRecordResponse,
)
def authentication_login_with_domain_and_creds_endpoint(
request: Request, data: EndpointBaseRequestModel
) -> Dict[str, Any]:
event_2_catch = AuthenticationLoginEventMethods.retrieve_event(
event_function_code=f"{template_event.key}"
)
data = event_2_catch.REQUEST_VALIDATOR(**data.data)
return event_2_catch.endpoint_callable(request=request, data=data)
class AccountCreateEventMethod(MethodToEvent):
event_type = "CREATE"
event_description = ""
event_category = ""
__event_keys__ = {
"31f4f32f-0cd4-4995-8a6a-f9f56335848a": "account_records_create",
}
__event_validation__ = {
"31f4f32f-0cd4-4995-8a6a-f9f56335848a": (
InsertAccountRecord,
[AccountRecords.__language_model__],
),
}
@classmethod
def account_records_create(
cls,
data: InsertAccountRecordRequestModel,
token_dict: Union[EmployeeTokenObject, OccupantTokenObject],
):
data_dict = data.excluded_dump()
if isinstance(token_dict, OccupantTokenObject):
db_session = AccountRecords.new_session()
build_iban = BuildIbans.filter_one(
BuildIbans.iban == data.iban,
BuildIbans.build_id == token_dict.selected_occupant.build_id,
db=db_session,
).data
if not build_iban:
raise BuildIbans.raise_http_exception(
status_code="HTTP_404_NOT_FOUND",
error_case="UNAUTHORIZED",
message=f"{data.iban} is not found in company related to your organization",
data={
"iban": data.iban,
},
)
account_record = AccountRecords.find_or_create(**data.excluded_dump())
return AlchemyJsonResponse(
completed=True,
message="Account record created successfully",
result=account_record,
)
elif isinstance(token_dict, EmployeeTokenObject):
# Build.pre_query = Build.select_action(
# employee_id=token_dict.selected_employee.employee_id,
# )
# build_ids_list = Build.filter_all(
# )
# build_iban = BuildIbans.filter_one(
# BuildIbans.iban == data.iban,
# BuildIbans.build_id.in_([build.id for build in build_ids_list.data]),
# ).data
# if not build_iban:
# BuildIbans.raise_http_exception(
# status_code="HTTP_404_NOT_FOUND",
# error_case="UNAUTHORIZED",
# message=f"{data.iban} is not found in company related to your organization",
# data={
# "iban": data.iban,
# },
# )
bank_date = system_arrow.get(data.bank_date)
data_dict["bank_date_w"] = bank_date.weekday()
data_dict["bank_date_m"] = bank_date.month
data_dict["bank_date_d"] = bank_date.day
data_dict["bank_date_y"] = bank_date.year
if int(data.currency_value) < 0:
debit_type = ApiEnumDropdown.filter_by_one(
system=True, enum_class="DebitTypes", key="DT-D"
).data
data_dict["receive_debit"] = debit_type.id
data_dict["receive_debit_uu_id"] = str(debit_type.uu_id)
else:
debit_type = ApiEnumDropdown.filter_by_one(
system=True, enum_class="DebitTypes", key="DT-R"
).data
data_dict["receive_debit"] = debit_type.id
data_dict["receive_debit_uu_id"] = str(debit_type.uu_id)
account_record = AccountRecords.insert_one(data_dict).data
return AlchemyJsonResponse(
completed=True,
message="Account record created successfully",
result=account_record,
)
class AccountUpdateEventMethod(MethodToEvent):
event_type = "UPDATE"
event_description = ""
event_category = ""
__event_keys__ = {
"ec98ef2c-bcd0-432d-a8f4-1822a56c33b2": "account_records_update",
}
__event_validation__ = {
"ec98ef2c-bcd0-432d-a8f4-1822a56c33b2": (
UpdateAccountRecord,
[AccountRecords.__language_model__],
),
}
@classmethod
def account_records_update(
cls,
build_uu_id: str,
data: UpdateAccountRecordRequestModel,
token_dict: Union[EmployeeTokenObject, OccupantTokenObject],
):
if isinstance(token_dict, OccupantTokenObject):
pass
elif isinstance(token_dict, EmployeeTokenObject):
pass
AccountRecords.build_parts_id = token_dict.selected_occupant.build_part_id
account_record = AccountRecords.update_one(build_uu_id, data).data
return AlchemyJsonResponse(
completed=True,
message="Account record updated successfully",
result=account_record,
cls_object=AccountRecords,
response_model=UpdateAccountRecord,
)
AuthenticationLoginEventMethods.endpoint_callable = (
authentication_login_with_domain_and_creds_endpoint
)

0
Events/AllEvents/events/account/api_events.py Normal file
View File

15
Events/AllEvents/events/account/cluster.py Normal file
View File

@ -0,0 +1,15 @@
from Events.Engine.abstract_class import CategoryCluster
# from info import template_page_info
AccountCluster = CategoryCluster(
name="AccountCluster",
tags=["template"],
prefix="/accounts",
description="Account Cluster",
pageinfo=None,
endpoints={},
include_in_schema=True,
sub_category=[],
)

350
Events/AllEvents/events/account/function_handlers.py Normal file
View File

@ -0,0 +1,350 @@
"""
Account records service implementation.
"""
from typing import Union
from pydantic import Field
from Events.base_request_model import TokenDictType, BaseRouteModel
from ApiLayers.ApiValidations.Custom.token_objects import (
OccupantTokenObject,
EmployeeTokenObject,
)
from ApiLayers.ApiLibrary import system_arrow
from ApiLayers.ApiValidations.Request.account_records import (
InsertAccountRecord,
UpdateAccountRecord,
)
from ApiLayers.ApiValidations.Request.base_validations import ListOptions
from ApiLayers.Schemas import (
BuildLivingSpace,
AccountRecords,
BuildIbans,
BuildDecisionBookPayments,
ApiEnumDropdown,
)
from Services.PostgresDb.Models.response import (
PostgresResponse,
)
from ApiLayers.ApiValidations.Response import AccountRecordResponse
from .models import (
InsertAccountRecordRequestModel,
UpdateAccountRecordRequestModel,
ListOptionsRequestModel,
)
class AccountListEventMethod(BaseRouteModel):
event_type = "SELECT"
event_description = ""
event_category = ""
__event_keys__ = {
"7192c2aa-5352-4e36-98b3-dafb7d036a3d": "account_records_list",
"208e6273-17ef-44f0-814a-8098f816b63a": "account_records_list_flt_res",
}
__event_validation__ = {
"7192c2aa-5352-4e36-98b3-dafb7d036a3d": (
AccountRecordResponse,
[AccountRecords.__language_model__],
),
"208e6273-17ef-44f0-814a-8098f816b63a": (
AccountRecordResponse,
[AccountRecords.__language_model__],
),
}
@classmethod
def account_records_list(
cls,
list_options: ListOptionsRequestModel,
token_dict: Union[EmployeeTokenObject, OccupantTokenObject],
):
db_session = AccountRecords.new_session()
if isinstance(token_dict, OccupantTokenObject):
AccountRecords.pre_query = AccountRecords.filter_all(
AccountRecords.company_id
== token_dict.selected_occupant.responsible_company_id,
db=db_session,
).query
elif isinstance(token_dict, EmployeeTokenObject):
AccountRecords.pre_query = AccountRecords.filter_all(
AccountRecords.company_id == token_dict.selected_company.company_id,
db=db_session,
).query
AccountRecords.filter_attr = list_options
records = AccountRecords.filter_all(db=db_session)
return AlchemyJsonResponse(
completed=True,
message="Account records listed successfully",
result=records,
cls_object=AccountRecords,
filter_attributes=list_options,
response_model=AccountRecordResponse,
)
@classmethod
def account_records_list_flt_res(
cls,
list_options: ListOptionsRequestModel,
token_dict: Union[EmployeeTokenObject, OccupantTokenObject],
):
db_session = AccountRecords.new_session()
if not isinstance(token_dict, OccupantTokenObject):
raise AccountRecords.raise_http_exception(
status_code="HTTP_404_NOT_FOUND",
error_case="UNAUTHORIZED",
message="Only Occupant can see this data",
data={},
)
return_list = []
living_space: BuildLivingSpace = BuildLivingSpace.filter_by_one(
id=token_dict.selected_occupant.living_space_id
).data
if not living_space:
raise AccountRecords.raise_http_exception(
status_code="HTTP_404_NOT_FOUND",
error_case="UNAUTHORIZED",
message="Living space not found",
data={},
)
if not list_options:
list_options = ListOptions()
main_filters = [
AccountRecords.living_space_id
== token_dict.selected_occupant.living_space_id,
BuildDecisionBookPayments.process_date
>= str(system_arrow.now().shift(months=-3).date()),
BuildDecisionBookPayments.process_date
< str(system_arrow.find_last_day_of_month(living_space.expiry_ends)),
BuildDecisionBookPayments.process_date
>= str(system_arrow.get(living_space.expiry_starts)),
BuildDecisionBookPayments.is_confirmed == True,
AccountRecords.active == True,
]
order_type = "desc"
if list_options.order_type:
order_type = "asc" if list_options.order_type[0] == "a" else "desc"
order_by_list = BuildDecisionBookPayments.process_date.desc()
if list_options.order_field:
if list_options.order_field == "process_date":
order_by_list = (
BuildDecisionBookPayments.process_date.asc()
if order_type == "asc"
else BuildDecisionBookPayments.process_date.desc()
)
if list_options.order_field == "bank_date":
order_by_list = (
AccountRecords.bank_date.desc()
if order_type == "asc"
else AccountRecords.bank_date.asc()
)
if list_options.order_field == "currency_value":
order_by_list = (
AccountRecords.currency_value.desc()
if order_type == "asc"
else AccountRecords.currency_value.asc()
)
if list_options.order_field == "process_comment":
order_by_list = (
AccountRecords.process_comment.desc()
if order_type == "asc"
else AccountRecords.process_comment.asc()
)
if list_options.order_field == "payment_amount":
order_by_list = (
BuildDecisionBookPayments.payment_amount.desc()
if order_type == "asc"
else BuildDecisionBookPayments.payment_amount.asc()
)
if list_options.query:
for key, value in list_options.query.items():
if key == "process_date":
main_filters.append(BuildDecisionBookPayments.process_date == value)
if key == "bank_date":
main_filters.append(AccountRecords.bank_date == value)
if key == "currency":
main_filters.append(BuildDecisionBookPayments.currency == value)
if key == "currency_value":
main_filters.append(AccountRecords.currency_value == value)
if key == "process_comment":
main_filters.append(AccountRecords.process_comment == value)
if key == "payment_amount":
main_filters.append(
BuildDecisionBookPayments.payment_amount == value
)
query = (
AccountRecords.session.query(
BuildDecisionBookPayments.process_date,
BuildDecisionBookPayments.payment_amount,
BuildDecisionBookPayments.currency,
AccountRecords.bank_date,
AccountRecords.currency_value,
AccountRecords.process_comment,
BuildDecisionBookPayments.uu_id,
)
.join(
AccountRecords,
AccountRecords.id == BuildDecisionBookPayments.account_records_id,
)
.filter(*main_filters)
).order_by(order_by_list)
query.limit(list_options.size or 5).offset(
(list_options.page or 1 - 1) * list_options.size or 5
)
for list_of_values in query.all() or []:
return_list.append(
{
"process_date": list_of_values[0],
"payment_amount": list_of_values[1],
"currency": list_of_values[2],
"bank_date": list_of_values[3],
"currency_value": list_of_values[4],
"process_comment": list_of_values[5],
}
)
return AlchemyJsonResponse(
completed=True,
message="Account records listed successfully",
result=return_list,
cls_object=AccountRecords,
filter_attributes=list_options,
response_model=AccountRecordResponse,
)
class AccountCreateEventMethod(BaseRouteModel):
event_type = "CREATE"
event_description = ""
event_category = ""
__event_keys__ = {
"31f4f32f-0cd4-4995-8a6a-f9f56335848a": "account_records_create",
}
__event_validation__ = {
"31f4f32f-0cd4-4995-8a6a-f9f56335848a": (
InsertAccountRecord,
[AccountRecords.__language_model__],
),
}
@classmethod
def account_records_create(
cls,
data: InsertAccountRecordRequestModel,
token_dict: Union[EmployeeTokenObject, OccupantTokenObject],
):
data_dict = data.excluded_dump()
if isinstance(token_dict, OccupantTokenObject):
db_session = AccountRecords.new_session()
build_iban = BuildIbans.filter_one(
BuildIbans.iban == data.iban,
BuildIbans.build_id == token_dict.selected_occupant.build_id,
db=db_session,
).data
if not build_iban:
raise BuildIbans.raise_http_exception(
status_code="HTTP_404_NOT_FOUND",
error_case="UNAUTHORIZED",
message=f"{data.iban} is not found in company related to your organization",
data={
"iban": data.iban,
},
)
account_record = AccountRecords.find_or_create(**data.excluded_dump())
return AlchemyJsonResponse(
completed=True,
message="Account record created successfully",
result=account_record,
)
elif isinstance(token_dict, EmployeeTokenObject):
# Build.pre_query = Build.select_action(
# employee_id=token_dict.selected_employee.employee_id,
# )
# build_ids_list = Build.filter_all(
# )
# build_iban = BuildIbans.filter_one(
# BuildIbans.iban == data.iban,
# BuildIbans.build_id.in_([build.id for build in build_ids_list.data]),
# ).data
# if not build_iban:
# BuildIbans.raise_http_exception(
# status_code="HTTP_404_NOT_FOUND",
# error_case="UNAUTHORIZED",
# message=f"{data.iban} is not found in company related to your organization",
# data={
# "iban": data.iban,
# },
# )
bank_date = system_arrow.get(data.bank_date)
data_dict["bank_date_w"] = bank_date.weekday()
data_dict["bank_date_m"] = bank_date.month
data_dict["bank_date_d"] = bank_date.day
data_dict["bank_date_y"] = bank_date.year
if int(data.currency_value) < 0:
debit_type = ApiEnumDropdown.filter_by_one(
system=True, enum_class="DebitTypes", key="DT-D"
).data
data_dict["receive_debit"] = debit_type.id
data_dict["receive_debit_uu_id"] = str(debit_type.uu_id)
else:
debit_type = ApiEnumDropdown.filter_by_one(
system=True, enum_class="DebitTypes", key="DT-R"
).data
data_dict["receive_debit"] = debit_type.id
data_dict["receive_debit_uu_id"] = str(debit_type.uu_id)
account_record = AccountRecords.insert_one(data_dict).data
return AlchemyJsonResponse(
completed=True,
message="Account record created successfully",
result=account_record,
)
class AccountUpdateEventMethod(BaseRouteModel):
event_type = "UPDATE"
event_description = ""
event_category = ""
__event_keys__ = {
"ec98ef2c-bcd0-432d-a8f4-1822a56c33b2": "account_records_update",
}
__event_validation__ = {
"ec98ef2c-bcd0-432d-a8f4-1822a56c33b2": (
UpdateAccountRecord,
[AccountRecords.__language_model__],
),
}
@classmethod
def account_records_update(
cls,
build_uu_id: str,
data: UpdateAccountRecordRequestModel,
token_dict: Union[EmployeeTokenObject, OccupantTokenObject],
):
if isinstance(token_dict, OccupantTokenObject):
pass
elif isinstance(token_dict, EmployeeTokenObject):
pass
AccountRecords.build_parts_id = token_dict.selected_occupant.build_part_id
account_record = AccountRecords.update_one(build_uu_id, data).data
return AlchemyJsonResponse(
completed=True,
message="Account record updated successfully",
result=account_record,
cls_object=AccountRecords,
response_model=UpdateAccountRecord,
)

4
Events/AllEvents/template/template/api_events.py
View File

@ -18,6 +18,4 @@ template_event = Event(
)
template_event.endpoint_callable = (
TemplateFunctions.template_example_function()
)
template_event.endpoint_callable = TemplateFunctions.template_example_function()

2
Events/AllEvents/template/template/info.py
View File

@ -7,5 +7,5 @@ template_page_info = PageInfo(
description={"en": "template"},
icon="",
parent="",
url=""
url="",
)

5
Events/AllEvents/template/template/models.py
View File

@ -1,12 +1,9 @@
from ApiLayers.ApiValidations.Request import (
BaseModelRegular
)
from ApiLayers.ApiValidations.Request import BaseModelRegular
class TemplateRequestModels:
TemplateRequestModelX = BaseModelRegular
class TemplateResponseModels:
TemplateResponseModelsX = BaseModelRegular

1
Events/AllEvents/validations/__init__.py
View File

@ -2,7 +2,6 @@
Validations package initialization.
"""
from .validation.cluster import ValidationsCluster

48
Events/AllEvents/validations/validation/function_handlers.py
View File

@ -5,7 +5,10 @@ Validation function handlers
from typing import Dict, Any
from fastapi import Request
from ApiLayers.AllConfigs.Redis.configs import RedisValidationKeysAction, RedisCategoryKeys
from ApiLayers.AllConfigs.Redis.configs import (
RedisValidationKeysAction,
RedisCategoryKeys,
)
from Services.Redis.Actions.actions import RedisActions
from Events.base_request_model import BaseRouteModel
@ -22,7 +25,7 @@ class ValidateBase:
@property
def function_codes(self):
redis_function_codes= RedisActions.get_json(
redis_function_codes = RedisActions.get_json(
list_keys=[f"{self.redis_key}{self.url}"]
)
if redis_function_codes.status:
@ -31,9 +34,13 @@ class ValidateBase:
@property
def intersection(self):
intersection = list(set(self.function_codes).intersection(set(self.reachable_codes)))
intersection = list(
set(self.function_codes).intersection(set(self.reachable_codes))
)
if not len(intersection) == 1:
raise ValueError("Users reachable function codes does not match or match more than one.")
raise ValueError(
"Users reachable function codes does not match or match more than one."
)
return intersection[0]
@ -44,9 +51,11 @@ class RedisHeaderRetrieve(ValidateBase):
@property
def header(self):
"""
Headers: Headers which is merged with response model && language models of event
Headers: Headers which is merged with response model && language models of event
"""
redis_header= RedisActions.get_json(list_keys=[f"{self.redis_key}:{self.intersection}"])
redis_header = RedisActions.get_json(
list_keys=[f"{self.redis_key}:{self.intersection}"]
)
if redis_header.status:
return redis_header.first
raise ValueError("Header not found")
@ -59,9 +68,11 @@ class RedisValidationRetrieve(ValidateBase):
@property
def validation(self):
"""
Validation: Validation of event which is merged with response model && language models of event
Validation: Validation of event which is merged with response model && language models of event
"""
redis_validation = RedisActions.get_json(list_keys=[f"{self.redis_key}:{self.intersection}"])
redis_validation = RedisActions.get_json(
list_keys=[f"{self.redis_key}:{self.intersection}"]
)
if redis_validation.status:
return redis_validation.first
raise ValueError("Header not found")
@ -72,8 +83,8 @@ class ValidationsBoth(RedisHeaderRetrieve, RedisValidationRetrieve):
@property
def both(self) -> Dict[str, Any]:
"""
Headers: Headers which is merged with response model && language models of event
Validation: Validation of event which is merged with response model && language models of event
Headers: Headers which is merged with response model && language models of event
Validation: Validation of event which is merged with response model && language models of event
"""
return {"headers": self.header, "validation": self.validation}
@ -85,14 +96,23 @@ class RetrieveValidation(BaseRouteModel):
"""
Retrieve validation by event function code
"""
if getattr(data, 'asked_field', "") not in ValidationsConfig.SUPPORTED_VALIDATIONS:
raise ValueError(f"Invalid asked field please retry with valid fields {ValidationsConfig.SUPPORTED_VALIDATIONS}")
if (
getattr(data, "asked_field", "")
not in ValidationsConfig.SUPPORTED_VALIDATIONS
):
raise ValueError(
f"Invalid asked field please retry with valid fields {ValidationsConfig.SUPPORTED_VALIDATIONS}"
)
reachable_codes = []
if cls.context_retriever.token.is_employee:
reachable_codes = cls.context_retriever.token.selected_company.reachable_event_codes
reachable_codes = (
cls.context_retriever.token.selected_company.reachable_event_codes
)
elif cls.context_retriever.token.is_occupant:
reachable_codes = cls.context_retriever.token.selected_occupant.reachable_event_codes
reachable_codes = (
cls.context_retriever.token.selected_occupant.reachable_event_codes
)
validate_dict = dict(url=data.url, reachable_code=reachable_codes)
if data.asked_field == "all":

2
Events/AllEvents/validations/validation/info.py
View File

@ -7,5 +7,5 @@ template_page_info = PageInfo(
description={"en": "template"},
icon="",
parent="",
url=""
url="",
)

1
Events/AllEvents/validations/validation/models.py
View File

@ -1,6 +1,7 @@
"""
Validation records request and response models.
"""
from typing import Optional
from pydantic import BaseModel

10
Events/AllEvents/validations/validation/validation.py
View File

@ -27,11 +27,15 @@ ValidationEventMethods = MethodToEvent(
def authentication_login_with_domain_and_creds_endpoint(
request: Request, data: EndpointBaseRequestModel
request: Request, data: EndpointBaseRequestModel
) -> Dict[str, Any]:
function = ValidationEventMethods.retrieve_event(event_function_code=f"{validation_event.key}")
function = ValidationEventMethods.retrieve_event(
event_function_code=f"{validation_event.key}"
)
data = function.REQUEST_VALIDATOR(**data.data)
RetrieveValidation.context_retriever = ContextRetrievers(func=authentication_login_with_domain_and_creds_endpoint)
RetrieveValidation.context_retriever = ContextRetrievers(
func=authentication_login_with_domain_and_creds_endpoint
)
return function.endpoint_callable(request=request, data=data)

18
Events/Engine/abstract_class.py
View File

@ -188,15 +188,15 @@ class CategoryCluster:
INCLUDE_IN_SCHEMA: Optional[bool] = True
def __init__(
self,
name: str,
tags: list,
prefix: str,
description: str,
endpoints: dict[str, MethodToEvent],
sub_category: list,
pageinfo: Optional[PageInfo] = None,
include_in_schema: Optional[bool] = True,
self,
name: str,
tags: list,
prefix: str,
description: str,
endpoints: dict[str, MethodToEvent],
sub_category: list,
pageinfo: Optional[PageInfo] = None,
include_in_schema: Optional[bool] = True,
):
self.NAME = name
self.TAGS = tags

12
Events/Engine/set_defaults/setClusters.py
View File

@ -88,11 +88,13 @@ class PrepareEvents(DecoratorModule):
cluster.get_redis_cluster_index_value()
)
# [SAVE]REDIS => CLUSTER_FUNCTION_CODES = {"ClusterToMethod"} : [FUNCTION_CODE, ...]}
self.valid_redis_items.CLUSTER_FUNCTION_CODES_VALUE.update({
f"{self.valid_redis_items.CLUSTER_FUNCTION_CODES_KEY}:{cluster.name}": tuple(
cluster.retrieve_all_function_codes()
)
})
self.valid_redis_items.CLUSTER_FUNCTION_CODES_VALUE.update(
{
f"{self.valid_redis_items.CLUSTER_FUNCTION_CODES_KEY}:{cluster.name}": tuple(
cluster.retrieve_all_function_codes()
)
}
)
for method_endpoint in list(cluster.ENDPOINTS.values()):
# [SAVE]REDIS => ENDPOINT2CLASS = {MethodEvent: Endpoint("/.../.../..."), ...}

10
Services/PostgresDb/Models/filter_functions.py
View File

@ -140,7 +140,9 @@ class QueryModel(ArgumentModel):
@classmethod
def filter_all_system(
cls: Type[T], *args: Union[BinaryExpression, ColumnExpressionArgument], db: Session
cls: Type[T],
*args: Union[BinaryExpression, ColumnExpressionArgument],
db: Session,
) -> PostgresResponse:
"""
Filter multiple records by expressions without status filtering.
@ -158,7 +160,11 @@ class QueryModel(ArgumentModel):
return PostgresResponse(pre_query=cls._query(db), query=query, is_array=True)
@classmethod
def filter_all(cls: Type[T], *args: Union[BinaryExpression, ColumnExpressionArgument], db: Session) -> PostgresResponse:
def filter_all(
cls: Type[T],
*args: Union[BinaryExpression, ColumnExpressionArgument],
db: Session,
) -> PostgresResponse:
"""
Filter multiple records by expressions.

44
Services/PostgresDb/Models/pagination.py
View File

@ -2,7 +2,8 @@ from __future__ import annotations
from typing import Any, Dict, Optional, Union
from sqlalchemy import desc, asc
from pydantic import BaseModel
from AllConfigs.SqlDatabase.configs import PaginateConfig
from ApiLayers.AllConfigs.SqlDatabase.configs import PaginateConfig
from ApiLayers.ApiValidations.Request import ListOptions
from Services.PostgresDb.Models.response import PostgresResponse
@ -183,3 +184,44 @@ class PaginationResult:
if self.response_type
else queried_data.get_dict()
)
class QueryOptions:
def __init__(self, table, data: Union[dict, ListOptions] = None, model_query: Optional[Any] = None):
self.table = table
self.data = data
self.model_query = model_query
if isinstance(data, dict):
self.data = ListOptions(**data)
self.validate_query()
if not self.data.order_type:
self.data.order_type = ["created_at"]
if not self.data.order_field:
self.data.order_field = ["uu_id"]
def validate_query(self):
if not self.data.query or not self.model_query:
return ()
cleaned_query, cleaned_query_by_model, last_dict = {}, {}, {}
for key, value in self.data.query.items():
cleaned_query[str(str(key).split("__")[0])] = value
cleaned_query_by_model[str(str(key).split("__")[0])] = (key, value)
cleaned_model = self.model_query(**cleaned_query)
for i in cleaned_query:
if hasattr(cleaned_model, i):
last_dict[str(cleaned_query_by_model[i][0])] = str(cleaned_query_by_model[i][1])
self.data.query = last_dict
def convert(self) -> tuple:
"""
self.table.convert(query)
(<sqlalchemy.sql.elements.BinaryExpression object at 0x7caaeacf0080>, <sqlalchemy.sql.elements.BinaryExpression object at 0x7caaea729b80>)
"""
if not self.data:
return ()
if not self.data.query:
return ()
print('query', self.data)
print('query', self.data.query)
return tuple(self.table.convert(self.data.query))

67
Services/PostgresDb/how_to.py
View File

@ -1,18 +1,26 @@
from Schemas import AddressNeighborhood
from typing import Optional
from ApiLayers.ApiValidations.Request import ListOptions
from ApiLayers.Schemas import AddressNeighborhood
from Services.PostgresDb.Models.crud_alchemy import Credentials
from Services.PostgresDb.Models.mixin import BasicMixin
from Services.PostgresDb.Models.pagination import Pagination, PaginationResult
from Services.PostgresDb.Models.pagination import Pagination, PaginationResult, QueryOptions
from pydantic import BaseModel
listing = False
listing = True
creating = False
updating = True
updating = False
new_session = AddressNeighborhood.new_session()
new_session_test = AddressNeighborhood.new_session()
BasicMixin.creds = Credentials(person_id=10, person_name="Berkay Super User")
class QueryModel(BaseModel):
neighborhood_name: Optional[str]
neighborhood_code: Optional[str]
if listing:
"""List Options and Queries"""
@ -20,25 +28,36 @@ if listing:
AddressNeighborhood.neighborhood_code.icontains("10"),
db=new_session,
).query
query_of_list_options = {
"neighborhood_name__ilike": "A%",
"neighborhood_code__contains": "3",
}
address_neighborhoods = AddressNeighborhood.filter_all(
*AddressNeighborhood.convert(query_of_list_options),
db=new_session,
)
pagination = Pagination(data=address_neighborhoods)
pagination.page = 9
pagination.size = 10
pagination.orderField = ["type_code", "neighborhood_code"]
pagination.orderType = ["desc", "asc"]
pagination_result = PaginationResult(
data=address_neighborhoods, pagination=pagination
)
print(pagination_result.pagination.as_dict())
print(pagination_result.data)
list_options = {
"page": 1,
"size": 11,
"order_field": ["type_code", "neighborhood_code"],
"order_type": ["asc", "desc"],
"query": {
"neighborhood_name__ilike": "A%",
"neighborhood_code__contains": "3",
"my_other_field__ilike": "B%",
"other_other_field__ilike": "C%",
}
}
query_options = QueryOptions(table=AddressNeighborhood, data=list_options, model_query=QueryModel)
address_neighborhoods = AddressNeighborhood.filter_all(*query_options.convert(), db=new_session)
pagination = Pagination(data=address_neighborhoods)
pagination.change(**list_options)
pagination_result = PaginationResult(data=address_neighborhoods, pagination=pagination)
print("as_dict", pagination_result.pagination.as_dict())
for i, row in enumerate(pagination_result.data):
print(i+1, row)
# list_options_valid = ListOptions(**list_options)
# pagination.page = 9
# pagination.size = 10
# pagination.orderField = ["type_code", "neighborhood_code"]
# pagination.orderType = ["asc", "asc"]
if creating:
"""Create Queries"""
@ -62,9 +81,7 @@ if creating:
if updating:
"""Update Queries"""
query_of_list_options = {
"uu_id": str("33a89767-d2dc-4531-8f66-7b650e22a8a7"),
}
query_of_list_options = {"uu_id": str("33a89767-d2dc-4531-8f66-7b650e22a8a7")}
print("query_of_list_options", query_of_list_options)
address_neighborhoods_one = AddressNeighborhood.filter_one(
*AddressNeighborhood.convert(query_of_list_options),