89 lines
2.6 KiB
Python
89 lines
2.6 KiB
Python
from typing import Optional, List
|
|
from sqlalchemy.orm import Session
|
|
from models.country import Country
|
|
from schemas.country import CountryCreate, CountryUpdate, CountrySchema
|
|
|
|
def get_country_by_name(db: Session, name: str) -> Optional[Country]:
|
|
"""
|
|
Retrieves a country from the database by its name.
|
|
|
|
Args:
|
|
db (Session): The database session.
|
|
name (str): The name of the country to retrieve.
|
|
|
|
Returns:
|
|
Optional[Country]: The country object if found, otherwise None.
|
|
"""
|
|
return db.query(Country).filter(Country.name == name).first()
|
|
|
|
def create_country(db: Session, country_data: CountryCreate) -> Country:
|
|
"""
|
|
Creates a new country in the database, including the continent.
|
|
|
|
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 get_all_countries(db: Session) -> List[CountrySchema]:
|
|
"""
|
|
Retrieves all countries from the database.
|
|
|
|
Args:
|
|
db (Session): The database session.
|
|
|
|
Returns:
|
|
List[CountrySchema]: A list of all country objects.
|
|
"""
|
|
countries = db.query(Country).all()
|
|
return [CountrySchema.from_orm(country) for country in countries]
|
|
|
|
def update_country(db: Session, country_id: str, country_data: CountryUpdate) -> Optional[CountrySchema]:
|
|
"""
|
|
Updates an existing country in the database, including the continent.
|
|
|
|
Args:
|
|
db (Session): The database session.
|
|
country_id (str): The ID of the country to update.
|
|
country_data (CountryUpdate): The updated data for the country.
|
|
|
|
Returns:
|
|
Optional[CountrySchema]: The updated country object if found, otherwise None.
|
|
"""
|
|
country = db.query(Country).filter(Country.id == country_id).first()
|
|
if not country:
|
|
return None
|
|
|
|
for field, value in country_data.dict(exclude_unset=True).items():
|
|
setattr(country, field, value)
|
|
|
|
db.commit()
|
|
db.refresh(country)
|
|
return CountrySchema.from_orm(country)
|
|
|
|
def delete_country(db: Session, country_id: str) -> bool:
|
|
"""
|
|
Deletes a country from the database.
|
|
|
|
Args:
|
|
db (Session): The database session.
|
|
country_id (str): The ID of the country to delete.
|
|
|
|
Returns:
|
|
bool: True if the country was deleted, False otherwise.
|
|
"""
|
|
country = db.query(Country).filter(Country.id == country_id).first()
|
|
if not country:
|
|
return False
|
|
|
|
db.delete(country)
|
|
db.commit()
|
|
return True |