49 lines
1.4 KiB
Python
49 lines
1.4 KiB
Python
from core.models.orm_models.Location import Location
|
|
from core.models.orm_models.Operator import Operator
|
|
|
|
from core.models.manage.session_management import get_session
|
|
from typing import Optional
|
|
from sqlalchemy import select
|
|
from sqlalchemy.orm import joinedload
|
|
|
|
class LocationController:
|
|
|
|
@staticmethod
|
|
def get(country_code: str, city_code: str):
|
|
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)
|