98 lines
2.7 KiB
Python
98 lines
2.7 KiB
Python
from typing import List, Optional, Dict, Union
|
|
from sqlalchemy.orm import Session
|
|
from sqlalchemy.exc import IntegrityError
|
|
import re
|
|
from models.country import Country
|
|
from schemas.country import CountryCreate, CountryUpdate
|
|
|
|
def validate_country_code(code: str) -> bool:
|
|
"""
|
|
Validate country code format (2 letter ISO code).
|
|
|
|
Args:
|
|
code: Two letter country code to validate
|
|
|
|
Returns:
|
|
bool: True if code format is valid, False otherwise
|
|
"""
|
|
return bool(re.match(r'^[A-Z]{2}$', code))
|
|
|
|
def get_country_by_code(db: Session, code: str) -> Optional[Country]:
|
|
"""
|
|
Get a country by its code.
|
|
|
|
Args:
|
|
db: Database session
|
|
code: Country code to search for
|
|
|
|
Returns:
|
|
Country object if found, None otherwise
|
|
"""
|
|
return db.query(Country).filter(Country.code == code.upper()).first()
|
|
|
|
def format_country_response(country: Country) -> Dict:
|
|
"""
|
|
Format country data for API response.
|
|
|
|
Args:
|
|
country: Country object to format
|
|
|
|
Returns:
|
|
Dict containing formatted country data
|
|
"""
|
|
return {
|
|
"name": country.name,
|
|
"code": country.code,
|
|
"currency": country.currency,
|
|
"capital": country.capital,
|
|
"region": country.region
|
|
}
|
|
|
|
def create_country_safely(db: Session, country_data: CountryCreate) -> Union[Country, Dict[str, str]]:
|
|
"""
|
|
Create a new country with validation and error handling.
|
|
|
|
Args:
|
|
db: Database session
|
|
country_data: Country data for creation
|
|
|
|
Returns:
|
|
Country object if created successfully, error dict otherwise
|
|
"""
|
|
if not validate_country_code(country_data.code):
|
|
return {"error": "Invalid country code format"}
|
|
|
|
existing_country = get_country_by_code(db, country_data.code)
|
|
if existing_country:
|
|
return {"error": "Country code already exists"}
|
|
|
|
try:
|
|
db_country = Country(
|
|
name=country_data.name,
|
|
code=country_data.code.upper(),
|
|
currency=country_data.currency,
|
|
capital=country_data.capital,
|
|
region=country_data.region
|
|
)
|
|
|
|
db.add(db_country)
|
|
db.commit()
|
|
db.refresh(db_country)
|
|
return db_country
|
|
|
|
except IntegrityError:
|
|
db.rollback()
|
|
return {"error": "Country name already exists"}
|
|
|
|
def get_countries_by_region(db: Session, region: str) -> List[Country]:
|
|
"""
|
|
Get all countries in a specific region.
|
|
|
|
Args:
|
|
db: Database session
|
|
region: Region to filter by
|
|
|
|
Returns:
|
|
List of Country objects in the specified region
|
|
"""
|
|
return db.query(Country).filter(Country.region == region).all() |