Big-Link-Man/src/database/interfaces.py

179 lines
4.9 KiB
Python

"""
Abstract repository interfaces for data access layer
"""
from abc import ABC, abstractmethod
from typing import Optional, List, Dict, Any
from src.database.models import User, SiteDeployment, Project, GeneratedContent
class IUserRepository(ABC):
"""Interface for User data access"""
@abstractmethod
def create(self, username: str, hashed_password: str, role: str) -> User:
"""Create a new user"""
pass
@abstractmethod
def get_by_id(self, user_id: int) -> Optional[User]:
"""Get a user by ID"""
pass
@abstractmethod
def get_by_username(self, username: str) -> Optional[User]:
"""Get a user by username"""
pass
@abstractmethod
def get_all(self) -> List[User]:
"""Get all users"""
pass
@abstractmethod
def update(self, user: User) -> User:
"""Update an existing user"""
pass
@abstractmethod
def delete(self, user_id: int) -> bool:
"""Delete a user by ID"""
pass
@abstractmethod
def exists(self, username: str) -> bool:
"""Check if a user exists by username"""
pass
class ISiteDeploymentRepository(ABC):
"""Interface for SiteDeployment data access"""
@abstractmethod
def create(
self,
site_name: str,
storage_zone_id: int,
storage_zone_name: str,
storage_zone_password: str,
storage_zone_region: str,
pull_zone_id: int,
pull_zone_bcdn_hostname: str,
custom_hostname: Optional[str] = None
) -> SiteDeployment:
"""Create a new site deployment"""
pass
@abstractmethod
def get_by_id(self, deployment_id: int) -> Optional[SiteDeployment]:
"""Get a site deployment by ID"""
pass
@abstractmethod
def get_by_hostname(self, custom_hostname: str) -> Optional[SiteDeployment]:
"""Get a site deployment by custom hostname"""
pass
@abstractmethod
def get_by_bcdn_hostname(self, bcdn_hostname: str) -> Optional[SiteDeployment]:
"""Get a site deployment by bunny.net CDN hostname"""
pass
@abstractmethod
def get_all(self) -> List[SiteDeployment]:
"""Get all site deployments"""
pass
@abstractmethod
def delete(self, deployment_id: int) -> bool:
"""Delete a site deployment by ID"""
pass
@abstractmethod
def exists(self, hostname: str) -> bool:
"""Check if a site deployment exists by either custom or bcdn hostname"""
pass
class IProjectRepository(ABC):
"""Interface for Project data access"""
@abstractmethod
def create(self, user_id: int, name: str, data: Dict[str, Any]) -> Project:
"""Create a new project"""
pass
@abstractmethod
def get_by_id(self, project_id: int) -> Optional[Project]:
"""Get a project by ID"""
pass
@abstractmethod
def get_by_user_id(self, user_id: int) -> List[Project]:
"""Get all projects for a user"""
pass
@abstractmethod
def get_all(self) -> List[Project]:
"""Get all projects"""
pass
@abstractmethod
def update(self, project: Project) -> Project:
"""Update an existing project"""
pass
@abstractmethod
def delete(self, project_id: int) -> bool:
"""Delete a project by ID"""
pass
class IGeneratedContentRepository(ABC):
"""Interface for GeneratedContent data access"""
@abstractmethod
def create(self, project_id: int, tier: int) -> GeneratedContent:
"""Create a new generated content record"""
pass
@abstractmethod
def get_by_id(self, content_id: int) -> Optional[GeneratedContent]:
"""Get generated content by ID"""
pass
@abstractmethod
def get_by_project_id(self, project_id: int) -> List[GeneratedContent]:
"""Get all generated content for a project"""
pass
@abstractmethod
def get_active_by_project(self, project_id: int, tier: int) -> Optional[GeneratedContent]:
"""Get the active generated content for a project/tier"""
pass
@abstractmethod
def get_by_tier(self, tier: int) -> List[GeneratedContent]:
"""Get all generated content for a specific tier"""
pass
@abstractmethod
def get_by_status(self, status: str) -> List[GeneratedContent]:
"""Get all generated content with a specific status"""
pass
@abstractmethod
def update(self, content: GeneratedContent) -> GeneratedContent:
"""Update an existing generated content record"""
pass
@abstractmethod
def set_active(self, content_id: int, project_id: int, tier: int) -> bool:
"""Set a content version as active (deactivates others)"""
pass
@abstractmethod
def delete(self, content_id: int) -> bool:
"""Delete a generated content record by ID"""
pass