add company domain, models, route

This commit is contained in:
2024-04-08 14:32:50 +00:00
parent 0f5a98b273
commit efdacfa0ce
12 changed files with 131 additions and 6 deletions

View File

View File

@@ -0,0 +1,10 @@
from api.domain import DomainError
class CompanyNotFoundError(DomainError): ...
class CompanyAlreadyExistsError(DomainError): ...
class UserAccessError(DomainError): ...

View File

@@ -0,0 +1,57 @@
import re
from dataclasses import dataclass
from uuid import UUID, uuid4
from api.domain import DomainValidationError
from api.domain.entity import DomainEntity
from api.domain.user.model import User, UserId
from api.domain.value_obj import DomainValueObject
@dataclass(frozen=True)
class CompanyEmail(DomainValueObject):
value: str
def __post_init__(self) -> None:
pattern = r"^[\w\.-]+@[a-zA-Z\d\.-]+\.[a-zA-Z]{2,}$"
if not re.match(pattern, self.value):
raise DomainValidationError(
"Invalid email format. Email must be in the format 'example@example.com'."
)
@dataclass(frozen=True)
class CompanyName(DomainValueObject):
value: str
def __post_init__(self) -> None:
if len(self.value) < 1:
raise DomainValidationError("First name must be at least 1 character long.")
if len(self.value) > 100:
raise DomainValidationError(
"First name must be at most 100 characters long."
)
if not self.value.isalpha():
raise DomainValidationError("First name must only contain letters.")
@dataclass(frozen=True)
class CompanyId(DomainValueObject):
value: UUID
@dataclass
class Company(DomainEntity[CompanyId]):
name: CompanyName
email: CompanyEmail
owner: User
@staticmethod
def create(name: str, email: str, owner: User) -> "Company":
return Company(
id=CompanyId(uuid4()),
name=CompanyName(name),
email=CompanyEmail(email),
owner=owner,
)

View File

@@ -0,0 +1,15 @@
from typing import Protocol
from api.domain.company.model import Company
from api.domain.user.model import User
class CompanyRepository(Protocol):
async def get_company(self, filter: dict) -> User | None:
raise NotImplementedError
async def create_company(self, company: Company) -> None:
raise NotImplementedError
async def get_workes_list(self) -> list[User] | None:
raise NotImplementedError