from abc import ABC from logging import Logger from fastapi import Depends, HTTPException from pydantic import UUID4, ConfigDict from sqlalchemy.orm import Session from mealie.core.config import get_app_dirs, get_app_settings from mealie.core.dependencies.dependencies import ( get_admin_user, get_current_user, get_integration_id, get_public_group, ) from mealie.core.exceptions import mealie_registered_exceptions from mealie.core.root_logger import get_logger from mealie.core.settings.directories import AppDirectories from mealie.core.settings.settings import AppSettings from mealie.db.db_setup import generate_session from mealie.lang import local_provider from mealie.lang.providers import Translator from mealie.repos._utils import NOT_SET, NotSet from mealie.repos.all_repositories import AllRepositories, get_repositories from mealie.routes._base.checks import OperationChecks from mealie.schema.household.household import HouseholdInDB from mealie.schema.user.user import GroupInDB, PrivateUser from mealie.services.event_bus_service.event_bus_service import EventBusService from mealie.services.event_bus_service.event_types import EventDocumentDataBase, EventTypes class _BaseController(ABC): # noqa: B024 session: Session = Depends(generate_session) translator: Translator = Depends(local_provider) _repos: AllRepositories | None = None _logger: Logger | None = None _settings: AppSettings | None = None _folders: AppDirectories | None = None @property def t(self): return self.translator.t if self.translator else local_provider().t @property def repos(self): if not self._repos: self._repos = AllRepositories(self.session, group_id=self.group_id, household_id=self.household_id) return self._repos @property def logger(self) -> Logger: if not self._logger: self._logger = get_logger() return self._logger @property def settings(self) -> AppSettings: if not self._settings: self._settings = get_app_settings() return self._settings @property def folders(self) -> AppDirectories: if not self._folders: self._folders = get_app_dirs() return self._folders @property def group_id(self) -> UUID4 | None | NotSet: return NOT_SET @property def household_id(self) -> UUID4 | None | NotSet: return NOT_SET model_config = ConfigDict(arbitrary_types_allowed=True) class BasePublicController(_BaseController): """ This is a public class for all User restricted controllers in the API. It includes the common SharedDependencies and some common methods used by all Admin controllers. """ ... class BasePublicGroupExploreController(BasePublicController): """ Base class for all controllers that are public and explore group data. """ group: GroupInDB = Depends(get_public_group) @property def group_id(self) -> UUID4 | None | NotSet: return self.group.id def get_public_household(self, household_slug_or_id: str | UUID4) -> HouseholdInDB: household = self.repos.households.get_by_slug_or_id(household_slug_or_id) if not household or household.preferences.private_household: raise HTTPException(404, "household not found") return household def get_explore_url_path(self, endpoint: str) -> str: if endpoint.startswith("/"): endpoint = endpoint[1:] return f"/explore/groups/{self.group.slug}/{endpoint}" class BasePublicHouseholdExploreController(BasePublicGroupExploreController): """ Base class for all controllers that are public and explore household data. """ @property def cross_household_repos(self): """ Household-level repos with no household filter. Public controllers don't have access to a household identifier; instead, they return all public data, filtered by the household preferences. When using this repo, the caller should filter by household preferences, e.g.: `household.preferences.privateHousehold = FALSE` """ return get_repositories(self.session, group_id=self.group_id, household_id=None) class BaseUserController(_BaseController): """ This is a base class for all User restricted controllers in the API. It includes the common SharedDependencies and some common methods used by all Admin controllers. """ user: PrivateUser = Depends(get_current_user) integration_id: str = Depends(get_integration_id) translator: Translator = Depends(local_provider) # Manual Cache _checks: OperationChecks def registered_exceptions(self, ex: type[Exception]) -> str: registered = { **mealie_registered_exceptions(self.translator), } return registered.get(ex, self.t("generic.server-error")) @property def group_id(self) -> UUID4: return self.user.group_id @property def household_id(self) -> UUID4: return self.user.household_id @property def group(self) -> GroupInDB: return self.repos.groups.get_one(self.group_id) @property def household(self) -> HouseholdInDB: return self.repos.households.get_one(self.household_id) @property def checks(self) -> OperationChecks: if not self._checks: self._checks = OperationChecks(self.user) return self._checks class BaseAdminController(BaseUserController): """ This is a base class for all Admin restricted controllers in the API. It includes the common Shared Dependencies and some common methods used by all Admin controllers. """ user: PrivateUser = Depends(get_admin_user) @property def repos(self): if not self._repos: # Admins have access to all groups and households, so we don't want to filter by group_id or household_id self._repos = AllRepositories(self.session, group_id=None, household_id=None) return self._repos class BaseCrudController(BaseUserController): """ Base class for all CRUD controllers to facilitate common CRUD functions. """ event_bus: EventBusService = Depends(EventBusService.as_dependency) def publish_event( self, event_type: EventTypes, document_data: EventDocumentDataBase, group_id: UUID4, household_id: UUID4 | None, message: str = "", ) -> None: self.event_bus.dispatch( integration_id=self.integration_id, group_id=group_id, household_id=household_id, event_type=event_type, document_data=document_data, message=message, )