62 lines
2 KiB
Python
62 lines
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 get_profile_location_data
|
|
|
|
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
|
|
|
|
|
|
# with get_session() as session:
|
|
# location_object = session.execute(
|
|
# select(Location)
|
|
# .where((Location.country_code == country_code) & (Location.code == city_code))
|
|
# .options(joinedload(Location.operator))
|
|
# ).scalar_one_or_none()
|
|
|
|
# return location_object
|
|
|
|
# legacy:
|
|
# Location.find(country_code, code)
|
|
|
|
|
|
|
|
@staticmethod
|
|
def get_all():
|
|
with get_session() as session:
|
|
all_records = session.execute(
|
|
select(Location)
|
|
.options(joinedload(Location.operator))
|
|
).scalars().all()
|
|
return all_records
|
|
|
|
# legacy:
|
|
# return Location.all()
|
|
|
|
|
|
# Deprecated legacy sync,
|
|
|
|
# from core.services.WebServiceApiService import WebServiceApiService
|
|
# @staticmethod
|
|
# def _sync(proxies: Optional[dict] = None):
|
|
|
|
# locations = WebServiceApiService.get_locations(proxies)
|
|
|
|
# Location.truncate()
|
|
# Location.save_many(locations)
|