35 lines
1.6 KiB
Python
35 lines
1.6 KiB
Python
from typing import Optional, List
|
|
from sqlalchemy.orm import Session
|
|
from sqlalchemy import select
|
|
from app.models.GeoModel import Country, State, City
|
|
from app.repositories.base_repository import BaseRepository
|
|
|
|
class GeographicRepository:
|
|
def get_countries(self, db: Session, active_only: bool = True) -> List[Country]:
|
|
stmt = select(Country).where(Country.deleted_at.is_(None))
|
|
if active_only:
|
|
stmt = stmt.where(Country.is_active.is_(True))
|
|
return list(db.execute(stmt).scalars().all())
|
|
|
|
def get_country_by_id(self, db: Session, country_id: int) -> Optional[Country]:
|
|
return db.get(Country, country_id)
|
|
|
|
def get_states_by_country(self, db: Session, country_id: int, active_only: bool = True) -> List[State]:
|
|
stmt = select(State).where(State.country_id == country_id, State.deleted_at.is_(None))
|
|
if active_only:
|
|
stmt = stmt.where(State.is_active.is_(True))
|
|
return list(db.execute(stmt).scalars().all())
|
|
|
|
def get_state_by_id(self, db: Session, state_id: int) -> Optional[State]:
|
|
return db.get(State, state_id)
|
|
|
|
def get_cities_by_state(self, db: Session, state_id: int, active_only: bool = True) -> List[City]:
|
|
stmt = select(City).where(City.state_id == state_id, City.deleted_at.is_(None))
|
|
if active_only:
|
|
stmt = stmt.where(City.is_active.is_(True))
|
|
return list(db.execute(stmt).scalars().all())
|
|
|
|
def get_city_by_id(self, db: Session, city_id: int) -> Optional[City]:
|
|
return db.get(City, city_id)
|
|
|
|
geo_repository = GeographicRepository()
|