84 lines
2.7 KiB
Python
84 lines
2.7 KiB
Python
from typing import List, Optional
|
|
from uuid import UUID
|
|
from sqlalchemy.orm import Session
|
|
from models.country import Country
|
|
from schemas.country import CountryCreate, CountryUpdate
|
|
|
|
def get_all_countries(db: Session) -> List[Country]:
|
|
"""
|
|
Retrieves all countries from the database.
|
|
|
|
Args:
|
|
db (Session): The database session.
|
|
|
|
Returns:
|
|
List[Country]: A list of all country objects.
|
|
"""
|
|
return db.query(Country).all()
|
|
|
|
def get_country_by_id(db: Session, country_id: UUID) -> Optional[Country]:
|
|
"""
|
|
Retrieves a single country by its ID.
|
|
|
|
Args:
|
|
db (Session): The database session.
|
|
country_id (UUID): The ID of the country to retrieve.
|
|
|
|
Returns:
|
|
Optional[Country]: The country object if found, otherwise None.
|
|
"""
|
|
return db.query(Country).filter(Country.id == country_id).first()
|
|
|
|
def create_country(db: Session, country_data: CountryCreate) -> Country:
|
|
"""
|
|
Creates a new country in the database.
|
|
|
|
Args:
|
|
db (Session): The database session.
|
|
country_data (CountryCreate): The data for the country to create.
|
|
|
|
Returns:
|
|
Country: The newly created country object.
|
|
"""
|
|
db_country = Country(**country_data.dict())
|
|
db.add(db_country)
|
|
db.commit()
|
|
db.refresh(db_country)
|
|
return db_country
|
|
|
|
def update_country(db: Session, country_id: UUID, country_data: CountryUpdate) -> Optional[Country]:
|
|
"""
|
|
Updates an existing country in the database.
|
|
|
|
Args:
|
|
db (Session): The database session.
|
|
country_id (UUID): The ID of the country to update.
|
|
country_data (CountryUpdate): The updated data for the country.
|
|
|
|
Returns:
|
|
Optional[Country]: The updated country object if found, otherwise None.
|
|
"""
|
|
db_country = db.query(Country).filter(Country.id == country_id).first()
|
|
if not db_country:
|
|
return None
|
|
|
|
for field, value in country_data.dict(exclude_unset=True).items():
|
|
setattr(db_country, field, value)
|
|
|
|
db.commit()
|
|
db.refresh(db_country)
|
|
return db_country
|
|
|
|
def get_european_countries(db: Session) -> List[Country]:
|
|
"""
|
|
Retrieves a list of European countries from the database.
|
|
|
|
Args:
|
|
db (Session): The database session.
|
|
|
|
Returns:
|
|
List[Country]: A list of country objects representing European countries.
|
|
"""
|
|
# Define a list of ISO codes for European countries
|
|
european_iso_codes = ["AL", "AD", "AT", "BY", "BE", "BA", "BG", "HR", "CY", "CZ", "DK", "EE", "FI", "FR", "DE", "GR", "HU", "IS", "IE", "IT", "LV", "LI", "LT", "LU", "MT", "MD", "MC", "ME", "NL", "NO", "PL", "PT", "RO", "RU", "SM", "RS", "SK", "SI", "ES", "SE", "CH", "UA", "GB", "VA"]
|
|
return db.query(Country).filter(Country.iso_code.in_(european_iso_codes)).all() |