from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from app.domain.entities.settings import CompanyProfile, GeneralSetting from app.domain.enums import ( CompanyBusinessType, CompanyCountry, CompanyIndustryType, CompanyTimeZone, DateDisplayFormat, FiscalYearStart, LandingPage, SystemCurrency, SystemLanguage, TimeDisplayFormat, ) from app.domain.repositories.settings_repository import SettingsRepository from app.infrastructure.db.models import CompanyProfileModel, GeneralSettingModel, new_id class MySQLSettingsRepository(SettingsRepository): def __init__(self, session: AsyncSession) -> None: self._session = session def _to_entity(self, row: GeneralSettingModel) -> GeneralSetting: return GeneralSetting( id=row.id, company_id=row.company_id, site_name=row.site_name, site_tagline=row.site_tagline, detail_description=row.detail_description, logo_path=row.logo_path, date_format=DateDisplayFormat(row.date_format), time_format=TimeDisplayFormat(row.time_format), currency=SystemCurrency(row.currency), items_per_page=row.items_per_page, default_landing_page=LandingPage(row.default_landing_page), system_language=SystemLanguage(row.system_language), enable_multi_warehouse=bool(row.enable_multi_warehouse), enable_barcode_scanning=bool(row.enable_barcode_scanning), enable_stock_alert=bool(row.enable_stock_alert), enable_batch_expiry=bool(row.enable_batch_expiry), allow_negative_stock=bool(row.allow_negative_stock), show_product_images=bool(row.show_product_images), created_at=row.created_at, updated_at=row.updated_at, ) async def get_general(self, company_id: str) -> GeneralSetting | None: result = await self._session.execute( select(GeneralSettingModel).where(GeneralSettingModel.company_id == company_id) ) row = result.scalar_one_or_none() return self._to_entity(row) if row else None async def get_latest_general(self) -> GeneralSetting | None: result = await self._session.execute( select(GeneralSettingModel) .order_by( GeneralSettingModel.logo_path.is_(None), GeneralSettingModel.updated_at.desc(), ) .limit(1) ) row = result.scalar_one_or_none() return self._to_entity(row) if row else None async def upsert_general(self, setting: GeneralSetting) -> GeneralSetting: result = await self._session.execute( select(GeneralSettingModel).where( GeneralSettingModel.company_id == setting.company_id ) ) row = result.scalar_one_or_none() if row is None: row = GeneralSettingModel( id=setting.id or new_id(), company_id=setting.company_id, created_at=setting.created_at, ) self._session.add(row) row.site_name = setting.site_name row.site_tagline = setting.site_tagline row.detail_description = setting.detail_description row.logo_path = setting.logo_path row.date_format = setting.date_format.value row.time_format = setting.time_format.value row.currency = setting.currency.value row.items_per_page = setting.items_per_page row.default_landing_page = setting.default_landing_page.value row.system_language = setting.system_language.value row.enable_multi_warehouse = setting.enable_multi_warehouse row.enable_barcode_scanning = setting.enable_barcode_scanning row.enable_stock_alert = setting.enable_stock_alert row.enable_batch_expiry = setting.enable_batch_expiry row.allow_negative_stock = setting.allow_negative_stock row.show_product_images = setting.show_product_images row.updated_at = setting.updated_at await self._session.commit() await self._session.refresh(row) return self._to_entity(row) def _to_profile(self, row: CompanyProfileModel) -> CompanyProfile: return CompanyProfile( id=row.id, company_id=row.company_id, company_name=row.company_name, date_of_establishment=row.date_of_establishment, company_tagline=row.company_tagline, business_type=( CompanyBusinessType(row.business_type) if row.business_type else None ), industry_type=( CompanyIndustryType(row.industry_type) if row.industry_type else None ), currency=SystemCurrency(row.currency), registration_number=row.registration_number, fiscal_year_start=FiscalYearStart(row.fiscal_year_start), tax_number=row.tax_number, time_zone=CompanyTimeZone(row.time_zone), website=row.website, default_language=SystemLanguage(row.default_language), logo_path=row.logo_path, address=row.address, country=CompanyCountry(row.country), state_province=row.state_province or "", city=row.city or "", postal_code=row.postal_code or "", phone=row.phone, alternate_phone=row.alternate_phone, email=row.email, alternate_email=row.alternate_email, fax=row.fax, number_of_employees=row.number_of_employees, description=row.description, notes=row.notes, facebook_url=row.facebook_url, twitter_url=row.twitter_url, linkedin_url=row.linkedin_url, instagram_url=row.instagram_url, youtube_url=row.youtube_url, whatsapp=row.whatsapp, created_at=row.created_at, updated_at=row.updated_at, ) async def get_company_profile(self, company_id: str) -> CompanyProfile | None: result = await self._session.execute( select(CompanyProfileModel).where(CompanyProfileModel.company_id == company_id) ) row = result.scalar_one_or_none() return self._to_profile(row) if row else None async def upsert_company_profile(self, profile: CompanyProfile) -> CompanyProfile: result = await self._session.execute( select(CompanyProfileModel).where( CompanyProfileModel.company_id == profile.company_id ) ) row = result.scalar_one_or_none() if row is None: row = CompanyProfileModel( id=profile.id or new_id(), company_id=profile.company_id, created_at=profile.created_at, ) self._session.add(row) row.company_name = profile.company_name row.date_of_establishment = profile.date_of_establishment row.company_tagline = profile.company_tagline row.business_type = profile.business_type.value if profile.business_type else None row.industry_type = profile.industry_type.value if profile.industry_type else None row.currency = profile.currency.value row.registration_number = profile.registration_number row.fiscal_year_start = profile.fiscal_year_start.value row.tax_number = profile.tax_number row.time_zone = profile.time_zone.value row.website = profile.website row.default_language = profile.default_language.value row.logo_path = profile.logo_path row.address = profile.address row.country = profile.country.value row.state_province = profile.state_province row.city = profile.city row.postal_code = profile.postal_code row.phone = profile.phone row.alternate_phone = profile.alternate_phone row.email = profile.email row.alternate_email = profile.alternate_email row.fax = profile.fax row.number_of_employees = profile.number_of_employees row.description = profile.description row.notes = profile.notes row.facebook_url = profile.facebook_url row.twitter_url = profile.twitter_url row.linkedin_url = profile.linkedin_url row.instagram_url = profile.instagram_url row.youtube_url = profile.youtube_url row.whatsapp = profile.whatsapp row.updated_at = profile.updated_at await self._session.commit() await self._session.refresh(row) return self._to_profile(row)