From bac4468d8a819b477159fe99f854dbe8e7e0c6d4 Mon Sep 17 00:00:00 2001 From: Backend IM Bot Date: Thu, 27 Mar 2025 16:15:54 +0000 Subject: [PATCH] Add helper functions for Country --- helpers/country_helpers.py | 98 ++++++++++++++++++++++++++++++++++++++ 1 file changed, 98 insertions(+) create mode 100644 helpers/country_helpers.py diff --git a/helpers/country_helpers.py b/helpers/country_helpers.py new file mode 100644 index 0000000..6619148 --- /dev/null +++ b/helpers/country_helpers.py @@ -0,0 +1,98 @@ +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() \ No newline at end of file