32 lines
1.2 KiB
Python
32 lines
1.2 KiB
Python
from core.models.orm_models.Location import Location
|
|
from core.models.orm_models.Operator import Operator
|
|
from core.errors.logger import logger
|
|
|
|
from core.models.manage.session_management import get_session
|
|
from typing import Optional
|
|
from sqlalchemy import select
|
|
from sqlalchemy.orm import joinedload
|
|
from core.models.DatabaseOperation import DatabaseOperation, DBErrorType
|
|
from core.models.orm_calls.location_calls import execute_location_sql
|
|
|
|
class LocationController:
|
|
|
|
@staticmethod
|
|
def get(country_code: str, city_code: str):
|
|
location_object = execute_location_sql(country_code, city_code)
|
|
if location_object.valid:
|
|
return location_object.data
|
|
else:
|
|
critical_error = f"[BaseProfile] Got invalid SQL Query which could not be solved by the wrapper, with error message {location_object.message} and type {location_object.error_type}"
|
|
logger.error(critical_error)
|
|
print(critical_error)
|
|
return None
|
|
|
|
@staticmethod
|
|
def get_all():
|
|
with get_session() as session:
|
|
all_records = session.execute(
|
|
select(Location)
|
|
.options(joinedload(Location.operator))
|
|
).scalars().all()
|
|
return all_records
|