pi3c
/
service_man
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

enum

main
Сергей Ванюшкин 2024-04-23 09:08:56 +00:00
parent ad92682eda
commit 76cf6950ae
16 changed files with 98 additions and 115 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
.pre-commit-config.yaml
View File

@ -23,7 +23,7 @@ repos:
rev: v3.15.2
hooks:
- id: pyupgrade
args: [ --py310-plus ]
args: [ --py3-plus ]
# Форматирует код под PEP8
- repo: https://github.com/pre-commit/mirrors-autopep8
@ -46,7 +46,7 @@ repos:
rev: 24.3.0
hooks:
- id: black
language_version: python3.10
language_version: python3
args: [ "--line-length=120" ]
# Проверка статических типов с помощью mypy

44
api/app_entrypoint/dependencies.py
View File

@ -1,6 +1,5 @@
from fastapi import FastAPI
from sqlalchemy.ext.asyncio import (AsyncEngine, AsyncSession,
async_sessionmaker)
from sqlalchemy.ext.asyncio import AsyncEngine, AsyncSession, async_sessionmaker
from api.application.abstractions.transaction import TransactionContextManager
from api.application.protocols.date_time import DateTimeProvider
@ -9,27 +8,36 @@ from api.application.protocols.password_hasher import PasswordHasher
from api.application.usecase.auth.auth_user import LoginUser
from api.application.usecase.auth.create_user import CreateUser
from api.application.usecase.company.create_company import CreateCompany
from api.application.usecase.company.get_users_company import \
GetCompaniesByOwnerEmail
from api.application.usecase.company.get_users_company import GetCompaniesByOwnerEmail
from api.domain.company.repository import CompanyRepository
from api.domain.user.repository import UserRepository
from api.infrastructure.auth.jwt_settings import JwtSettings
from api.infrastructure.dependencies.adapters import (create_engine,
create_session_maker,
get_transaction_context,
new_session)
from api.infrastructure.dependencies.configs import (app_settings,
get_db_settings,
get_jwt_settings)
from api.infrastructure.dependencies.protocols import (get_date_time_provider,
get_jwt_token_processor,
get_password_hasher,
get_user_login)
from api.infrastructure.dependencies.adapters import (
create_engine,
create_session_maker,
get_transaction_context,
new_session,
)
from api.infrastructure.dependencies.configs import (
app_settings,
get_db_settings,
get_jwt_settings,
)
from api.infrastructure.dependencies.protocols import (
get_date_time_provider,
get_jwt_token_processor,
get_password_hasher,
get_user_login,
)
from api.infrastructure.dependencies.repositories import (
get_company_repository, get_user_repository)
get_company_repository,
get_user_repository,
)
from api.infrastructure.dependencies.usecases import (
provide_create_company, provide_create_user,
provide_get_companies_by_email)
provide_create_company,
provide_create_user,
provide_get_companies_by_email,
)
from api.infrastructure.persistence.db_setings import DBSettings
from api.infrastructure.settings import Settings

40
api/app_entrypoint/error_handlers.py
View File

@ -2,27 +2,23 @@ from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
from api.domain.error import DomainValidationError
from api.domain.user.error import (UserAlreadyExistsError,
UserInvalidCredentialsError,
UserIsNotAuthorizedError)
from api.domain.user.error import (
UserAlreadyExistsError,
UserInvalidCredentialsError,
UserIsNotAuthorizedError,
)
from api.infrastructure.persistence.error import TransactionContextManagerError
async def transaction_error_exec_handler(
request: Request, exc: TransactionContextManagerError
) -> JSONResponse:
async def transaction_error_exec_handler(request: Request, exc: TransactionContextManagerError) -> JSONResponse:
return JSONResponse(status_code=400, content={"detail": exc.message})
async def validation_error_exc_handler(
request: Request, exc: DomainValidationError
) -> JSONResponse:
async def validation_error_exc_handler(request: Request, exc: DomainValidationError) -> JSONResponse:
return JSONResponse(status_code=400, content={"detail": exc.message})
async def user_authentication_error_exc_handler(
request: Request, exc: UserIsNotAuthorizedError
) -> JSONResponse:
async def user_authentication_error_exc_handler(request: Request, exc: UserIsNotAuthorizedError) -> JSONResponse:
return JSONResponse(
status_code=401,
content={"detail": exc.message},
@ -30,9 +26,7 @@ async def user_authentication_error_exc_handler(
)
async def user_already_exist_error_exc_handler(
request: Request, exc: UserAlreadyExistsError
) -> JSONResponse:
async def user_already_exist_error_exc_handler(request: Request, exc: UserAlreadyExistsError) -> JSONResponse:
return JSONResponse(status_code=409, content={"detail": exc.message})
@ -47,16 +41,8 @@ def init_exc_handlers(app: FastAPI) -> None:
DomainValidationError,
validation_error_exc_handler,
)
app.add_exception_handler(
UserIsNotAuthorizedError, user_authentication_error_exc_handler
)
app.add_exception_handler(
UserAlreadyExistsError, user_already_exist_error_exc_handler
)
app.add_exception_handler(
UserInvalidCredentialsError, user_invalid_credentials_error_exc_handler
)
app.add_exception_handler(UserIsNotAuthorizedError, user_authentication_error_exc_handler)
app.add_exception_handler(UserAlreadyExistsError, user_already_exist_error_exc_handler)
app.add_exception_handler(UserInvalidCredentialsError, user_invalid_credentials_error_exc_handler)
app.add_exception_handler(
TransactionContextManagerError, transaction_error_exec_handler
)
app.add_exception_handler(TransactionContextManagerError, transaction_error_exec_handler)

8
api/app_entrypoint/main.py
View File

@ -8,9 +8,11 @@ from api.app_entrypoint.dependencies import init_dependencies
from api.app_entrypoint.error_handlers import init_exc_handlers
from api.infrastructure.auth.jwt_settings import JwtSettings
from api.infrastructure.dependencies.adapters import create_engine
from api.infrastructure.dependencies.configs import (app_settings,
get_db_settings,
get_jwt_settings)
from api.infrastructure.dependencies.configs import (
app_settings,
get_db_settings,
get_jwt_settings,
)
from api.infrastructure.persistence.db_setings import DBSettings
from api.infrastructure.persistence.models import Base
from api.infrastructure.settings import Settings

15
api/application/contracts/user/roles.py Normal file
View File

@ -0,0 +1,15 @@
from enum import Enum
class UserRole(Enum):
# base roles
ADMIN = "Administrator"
# service provider roles
SUPPLIER_EMPLOYER = "Director"
SUPPLIER = "Supplie of service"
SUPPLIER_EMPLOYEE = "Supplie employee"
CLIENT = "Client"
EMPLOYER = "Client company employer"
EMPLOYEE = "Client company employee"

3
api/application/usecase/company/create_company.py
View File

@ -1,7 +1,6 @@
from api.application.abstractions.transaction import TransactionContextManager
from api.application.contracts.company.company_request import CreateNewCompany
from api.application.contracts.company.company_response import \
CompanyBaseResponse
from api.application.contracts.company.company_response import CompanyBaseResponse
from api.domain.company.repository import CompanyRepository

15
api/application/usecase/company/get_users_company.py
View File

@ -1,8 +1,6 @@
from api.application.abstractions.transaction import TransactionContextManager
from api.application.contracts.company.company_request import \
CompanyByOwnerEmail
from api.application.contracts.company.company_response import \
CompanyBaseResponse
from api.application.contracts.company.company_request import CompanyByOwnerEmail
from api.application.contracts.company.company_response import CompanyBaseResponse
from api.domain.company.repository import CompanyRepository
@ -17,10 +15,5 @@ class GetCompaniesByOwnerEmail:
async def execute(self, request: CompanyByOwnerEmail) -> list[CompanyBaseResponse]:
async with self.transaction:
companies = await self.company_repository.get_companies_by_owner_email(
filter={"email": request.email}
)
return [
CompanyBaseResponse(name=comp.name.value, email=comp.email.value)
for comp in companies
]
companies = await self.company_repository.get_companies_by_owner_email(filter={"email": request.email})
return [CompanyBaseResponse(name=comp.name.value, email=comp.email.value) for comp in companies]

4
api/application/usecase/user/get_user_by_email.py
View File

@ -4,9 +4,7 @@ from api.domain.user.repository import UserRepository
class GetUserByEmail:
def __init__(
self, transaction: TransactionContextManager, user_repository: UserRepository
) -> None:
def __init__(self, transaction: TransactionContextManager, user_repository: UserRepository) -> None:
self.transaction = transaction
self.user_repository = user_repository

8
api/domain/company/model.py
View File

@ -16,9 +16,7 @@ class CompanyEmail(DomainValueObject):
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'."
)
raise DomainValidationError("Invalid email format. Email must be in the format 'example@example.com'.")
@dataclass(frozen=True)
@ -29,9 +27,7 @@ class CompanyName(DomainValueObject):
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."
)
raise DomainValidationError("First name must be at most 100 characters long.")
if not self.value.isalpha():
raise DomainValidationError("First name must only contain letters.")

12
api/domain/user/model.py
View File

@ -15,9 +15,7 @@ class UserEmail(DomainValueObject):
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'."
)
raise DomainValidationError("Invalid email format. Email must be in the format 'example@example.com'.")
@dataclass(frozen=True)
@ -28,9 +26,7 @@ class UserFirstName(DomainValueObject):
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."
)
raise DomainValidationError("First name must be at most 100 characters long.")
if not self.value.isalpha():
raise DomainValidationError("First name must only contain letters.")
@ -43,9 +39,7 @@ class UserLastName(DomainValueObject):
if len(self.value) < 1:
raise DomainValidationError("Last name must be at least 1 character long.")
if len(self.value) > 100:
raise DomainValidationError(
"Last name must be at most 100 characters long."
)
raise DomainValidationError("Last name must be at most 100 characters long.")
if not self.value.isalpha():
raise DomainValidationError("Last name must only contain letters.")

17
api/infrastructure/dependencies/adapters.py
View File

@ -2,19 +2,22 @@ from collections.abc import AsyncIterable
from typing import Annotated
from fastapi import Depends
from sqlalchemy.ext.asyncio import (AsyncEngine, AsyncSession,
async_sessionmaker, create_async_engine)
from sqlalchemy.ext.asyncio import (
AsyncEngine,
AsyncSession,
async_sessionmaker,
create_async_engine,
)
from api.application.abstractions.transaction import TransactionContextManager
from api.infrastructure.dependencies.stub import Stub
from api.infrastructure.persistence.transaction import \
SqlalchemyTransactionContextManager
from api.infrastructure.persistence.transaction import (
SqlalchemyTransactionContextManager,
)
from api.infrastructure.settings import Settings
def get_transaction_context(
session: Annotated[AsyncSession, Depends(Stub(AsyncSession))]
) -> TransactionContextManager:
def get_transaction_context(session: Annotated[AsyncSession, Depends(Stub(AsyncSession))]) -> TransactionContextManager:
return SqlalchemyTransactionContextManager(session)

19
api/infrastructure/dependencies/usecases.py
View File

@ -6,8 +6,7 @@ from api.application.abstractions.transaction import TransactionContextManager
from api.application.protocols.password_hasher import PasswordHasher
from api.application.usecase.auth.create_user import CreateUser
from api.application.usecase.company.create_company import CreateCompany
from api.application.usecase.company.get_users_company import \
GetCompaniesByOwnerEmail
from api.application.usecase.company.get_users_company import GetCompaniesByOwnerEmail
from api.domain.company.repository import CompanyRepository
from api.domain.user.repository import UserRepository
from api.infrastructure.dependencies.stub import Stub
@ -15,9 +14,7 @@ from api.infrastructure.dependencies.stub import Stub
def provide_create_user(
user_repository: Annotated[UserRepository, Depends(Stub(UserRepository))],
transaction: Annotated[
TransactionContextManager, Depends(Stub(TransactionContextManager))
],
transaction: Annotated[TransactionContextManager, Depends(Stub(TransactionContextManager))],
password_hasher: Annotated[PasswordHasher, Depends(Stub(PasswordHasher))],
) -> CreateUser:
return CreateUser(
@ -29,19 +26,13 @@ def provide_create_user(
def provide_get_companies_by_email(
company_repository: Annotated[CompanyRepository, Depends(Stub(CompanyRepository))],
transaction: Annotated[
TransactionContextManager, Depends(Stub(TransactionContextManager))
],
transaction: Annotated[TransactionContextManager, Depends(Stub(TransactionContextManager))],
) -> GetCompaniesByOwnerEmail:
return GetCompaniesByOwnerEmail(
transaction=transaction, company_repository=company_repository
)
return GetCompaniesByOwnerEmail(transaction=transaction, company_repository=company_repository)
def provide_create_company(
company_repository: Annotated[CompanyRepository, Depends(Stub(CompanyRepository))],
transaction: Annotated[
TransactionContextManager, Depends(Stub(TransactionContextManager))
],
transaction: Annotated[TransactionContextManager, Depends(Stub(TransactionContextManager))],
) -> CreateCompany:
return CreateCompany(transaction=transaction, company_repository=company_repository)

4
api/infrastructure/persistence/models/company.py
View File

@ -30,9 +30,7 @@ class DepartmentModel(Base):
)
name: Mapped[str]
address: Mapped[str]
company_id: Mapped[uuid.UUID] = mapped_column(
ForeignKey("company.id", ondelete="CASCADE")
)
company_id: Mapped[uuid.UUID] = mapped_column(ForeignKey("company.id", ondelete="CASCADE"))
class CompanyDepartmentModel(Base):

9
api/infrastructure/persistence/repositories/company_repository.py
View File

@ -1,8 +1,13 @@
from sqlalchemy import text
from sqlalchemy.ext.asyncio import AsyncSession
from api.domain.company.model import (Company, CompanyAddress, CompanyEmail,
CompanyId, CompanyName)
from api.domain.company.model import (
Company,
CompanyAddress,
CompanyEmail,
CompanyId,
CompanyName,
)
from api.domain.company.repository import CompanyRepository
from api.domain.user.model import UserId

3
api/infrastructure/persistence/repositories/user_repository.py
View File

@ -2,8 +2,7 @@ from sqlalchemy import text
from sqlalchemy.ext.asyncio import AsyncSession
from api.domain.user import User, UserRepository
from api.domain.user.model import (UserEmail, UserFirstName, UserId,
UserLastName)
from api.domain.user.model import UserEmail, UserFirstName, UserId, UserLastName
class SqlAlchemyUserRepository(UserRepository):

8
api/presentation/routers/company.py
View File

@ -28,16 +28,12 @@ company_router = APIRouter(
async def get_my_companies(
request: Request,
token_processor: Annotated[JwtTokenProcessor, Depends(Stub(JwtTokenProcessor))],
usecase: Annotated[
GetCompaniesByOwnerEmail, Depends(Stub(GetCompaniesByOwnerEmail))
],
usecase: Annotated[GetCompaniesByOwnerEmail, Depends(Stub(GetCompaniesByOwnerEmail))],
) -> list[CompanyBaseResponse]:
token_data = token_processor.validate_token(request.scope["auth"])
if not token_data:
raise UserValidationError("Login required")
companies = await usecase.execute(
request=CompanyByOwnerEmail(email=token_data[1].value)
)
companies = await usecase.execute(request=CompanyByOwnerEmail(email=token_data[1].value))
return [
CompanyBaseResponse(
name=c.name,