added CRUD status, type and task models

parent 424eec46
from fastapi import APIRouter
# from .director.routes import router as director_route
# from .genre.routes import router as genre_route
# from .movie.routes import router as movie_route
from .status.routes import router as status_route
from .type.routes import router as type_route
from .task.routes import router as task_route
router = APIRouter(prefix='/api')
# router.include_router(director_route)
# router.include_router(genre_route)
# router.include_router(movie_route)
router.include_router(status_route)
router.include_router(type_route)
router.include_router(task_route)
from uuid import UUID
from starlette import status
import fastapi
from tortoise.exceptions import ValidationError
from base_ctrl import BaseController
from exceptions import common as common_exc, http as http_exc
from .services import StatusService
class StatusController(BaseController):
status_service = StatusService()
async def get_list(self, **kwargs):
return await self.status_service.get_status_list(**kwargs)
async def get(self, id: UUID):
try:
return await self.status_service.get_status(id)
except common_exc.NotFoundException as e:
raise http_exc.HTTPNotFoundException(detail=str(e))
async def create(self, **kwargs):
try:
return await self.status_service.create_status(**kwargs)
except common_exc.CreateException as e:
raise http_exc.HTTPBadRequestException(detail=str(e))
except ValidationError as e:
raise http_exc.HTTPBadRequestException(detail=f"Validation error: {str(e)}")
async def update(self, id: UUID, **kwargs):
try:
return await self.status_service.update_status(id, **kwargs)
except common_exc.UpdateException as e:
raise http_exc.HTTPBadRequestException(detail=str(e))
async def delete(self, id: UUID):
try:
await self.status_service.delete_status(id)
return fastapi.responses.Response(status_code=status.HTTP_204_NO_CONTENT)
except common_exc.DeleteException as e:
raise http_exc.HTTPBadRequestException(detail=str(e))
except common_exc.NotFoundException as e:
raise http_exc.HTTPNotFoundException(detail=str(e))
from uuid import UUID
import fastapi
from .schemas import StatusBaseSchema
from .ctrl import StatusController
router = fastapi.APIRouter(prefix='/statuses', tags=['Status'])
ctrl = StatusController()
@router.get('')
async def get_statuses(query: StatusBaseSchema = fastapi.Depends()):
return await ctrl.get_list(**query.model_dump(exclude_none=True))
@router.get('/{id}')
async def get_status(id: UUID):
return await ctrl.get(id)
@router.post('')
async def create_status(body: StatusBaseSchema):
return await ctrl.create(**body.model_dump(exclude_none=True))
@router.patch('/{id}')
async def update_status(id: UUID, body: StatusBaseSchema):
return await ctrl.update(id, **body.model_dump(exclude_none=True))
@router.delete('/{id}')
async def delete_status(id: UUID):
return await ctrl.delete(id)
from uuid import UUID
from pydantic import BaseModel, ConfigDict
class StatusBaseSchema(BaseModel):
name: str | None = None
class StatusIdSchema(BaseModel):
id: UUID
model_config = ConfigDict(from_attributes=True)
class StatusSchema(StatusBaseSchema, StatusIdSchema):
pass
from uuid import UUID
from db.repositories import StatusRepository
from .schemas import StatusSchema
class StatusService:
status_repository = StatusRepository()
async def get_status_list(self, **kwargs):
statuses = await self.status_repository.get_list(**kwargs)
return [StatusSchema.model_validate(status) for status in statuses]
async def get_status(self, id: UUID):
status = await self.status_repository.get(id)
return StatusSchema.model_validate(status)
async def create_status(self, **kwargs):
status = await self.status_repository.create(**kwargs)
return StatusSchema.model_validate(status)
async def update_status(self, id: UUID, **kwargs):
status = await self.status_repository.update(id, **kwargs)
return StatusSchema.model_validate(status)
async def delete_status(self, id: UUID):
return await self.status_repository.delete(id)
from uuid import UUID
from starlette import status
import fastapi
from tortoise.exceptions import ValidationError
from base_ctrl import BaseController
from exceptions import common as common_exc, http as http_exc
from .services import TaskService
class TaskController(BaseController):
task_service = TaskService()
async def get_list(self, **kwargs):
return await self.task_service.get_task_list(**kwargs)
async def get(self, id: UUID):
try:
return await self.task_service.get_task(id)
except common_exc.NotFoundException as e:
raise http_exc.HTTPNotFoundException(detail=str(e))
async def create(self, **kwargs):
try:
return await self.task_service.create_task(**kwargs)
except common_exc.CreateException as e:
raise http_exc.HTTPBadRequestException(detail=str(e))
except ValidationError as e:
raise http_exc.HTTPBadRequestException(detail=f"Validation error: {str(e)}")
async def update(self, id: UUID, **kwargs):
try:
return await self.task_service.update_task(id, **kwargs)
except common_exc.UpdateException as e:
raise http_exc.HTTPBadRequestException(detail=str(e))
async def delete(self, id: UUID):
try:
await self.task_service.delete_task(id)
return fastapi.responses.Response(status_code=status.HTTP_204_NO_CONTENT)
except common_exc.DeleteException as e:
raise http_exc.HTTPBadRequestException(detail=str(e))
except common_exc.NotFoundException as e:
raise http_exc.HTTPNotFoundException(detail=str(e))
from uuid import UUID
import fastapi
from .schemas import (
TaskBaseSchema, TaskPostSchema, TaskPatchSchema
)
from .ctrl import TaskController
router = fastapi.APIRouter(prefix='/tasks', tags=['Task'])
ctrl = TaskController()
@router.get('')
async def get_tasks(query: TaskBaseSchema = fastapi.Depends()):
return await ctrl.get_list(**query.model_dump(exclude_none=True))
@router.get('/{id}')
async def get_task(id: UUID):
return await ctrl.get(id)
@router.post('')
async def create_task(body: TaskPostSchema):
return await ctrl.create(**body.model_dump(exclude_none=True))
@router.patch('/{id}')
async def update_task(id: UUID, body: TaskPatchSchema):
return await ctrl.update(id, **body.model_dump(exclude_none=True))
@router.delete('/{id}')
async def delete_task(id: UUID):
return await ctrl.delete(id)
from uuid import UUID
from typing import List
from pydantic import BaseModel, ConfigDict
class TaskBaseSchema(BaseModel):
title: str | None = None
class TaskIdSchema(BaseModel):
id: UUID
model_config = ConfigDict(from_attributes=True)
class TaskListSchema(TaskBaseSchema, TaskIdSchema):
pass
class TaskPostSchema(TaskBaseSchema):
description: str
status: UUID | None = None
types: List[UUID] | None = None
class TaskPatchSchema(TaskBaseSchema):
description: str | None = None
status: UUID | None
types: List[UUID] | None = None
from uuid import UUID
from db.repositories import TaskRepository, StatusRepository
from exceptions import common as common_exc, http as http_exc
from db.models import Type, Status
from .schemas import TaskListSchema
class TaskService:
task_repository = TaskRepository()
status_repository = StatusRepository()
async def get_task_list(self, **kwargs):
tasks = await self.task_repository.get_list(**kwargs)
return [TaskListSchema.model_validate(task) for task in tasks]
async def get_task(self, id: UUID):
task = await self.task_repository.get(id=id)
await task.fetch_related('status', 'types')
status_id = task.status.id if task.status else None
types_data = [type.id for type in task.types]
task_data = {
"id": task.id,
"title": task.title,
"description": task.description,
"status": status_id,
"types": types_data
}
return task_data
async def create_task(self, **kwargs):
status_uuid = kwargs.get('status')
type_uuids = kwargs.get('types', [])
if status_uuid is None:
raise http_exc.HTTPBadRequestException(detail="Validation error: status: Value must not be None")
try:
status = await self.status_repository.get(status_uuid)
except common_exc.NotFoundException:
raise http_exc.HTTPBadRequestException(detail=f"status does not exist")
task = await self.task_repository.create(
title=kwargs.get('title'),
description=kwargs.get('description'),
status=status
)
types = await Type.filter(id__in=type_uuids)
await task.types.add(*types)
return await self.get_task(id=task.id)
async def update_task(self, id: UUID, **kwargs):
status_uuid = kwargs.get('status')
type_uuids = kwargs.get('types', [])
if status_uuid:
kwargs['status'] = await self.status_repository.get(status_uuid)
types = kwargs.pop('types', [])
task = await self.task_repository.update(id, **kwargs)
if type_uuids:
types = await Type.filter(id__in=types)
await task.types.clear()
await task.types.add(*types)
return await self.get_task(id=task.id)
async def delete_task(self, id: UUID):
return await self.task_repository.delete(id)
from uuid import UUID
from starlette import status
import fastapi
from tortoise.exceptions import ValidationError
from base_ctrl import BaseController
from exceptions import common as common_exc, http as http_exc
from .services import TypeService
class TypeController(BaseController):
type_service = TypeService()
async def get_list(self, **kwargs):
return await self.type_service.get_type_list(**kwargs)
async def get(self, id: UUID):
try:
return await self.type_service.get_type(id)
except common_exc.NotFoundException as e:
raise http_exc.HTTPNotFoundException(detail=str(e))
async def create(self, **kwargs):
try:
return await self.type_service.create_type(**kwargs)
except common_exc.CreateException as e:
raise http_exc.HTTPBadRequestException(detail=str(e))
except ValidationError as e:
raise http_exc.HTTPBadRequestException(detail=f"Validation error: {str(e)}")
async def update(self, id: UUID, **kwargs):
try:
return await self.type_service.update_type(id, **kwargs)
except common_exc.UpdateException as e:
raise http_exc.HTTPBadRequestException(detail=str(e))
async def delete(self, id: UUID):
try:
await self.type_service.delete_type(id)
return fastapi.responses.Response(status_code=status.HTTP_204_NO_CONTENT)
except common_exc.DeleteException as e:
raise http_exc.HTTPBadRequestException(detail=str(e))
except common_exc.NotFoundException as e:
raise http_exc.HTTPNotFoundException(detail=str(e))
from uuid import UUID
import fastapi
from .schemas import TypeBaseSchema
from .ctrl import TypeController
router = fastapi.APIRouter(prefix='/types', tags=['Type'])
ctrl = TypeController()
@router.get('')
async def get_types(query: TypeBaseSchema = fastapi.Depends()):
return await ctrl.get_list(**query.model_dump(exclude_none=True))
@router.get('/{id}')
async def get_type(id: UUID):
return await ctrl.get(id)
@router.post('')
async def create_type(body: TypeBaseSchema):
return await ctrl.create(**body.model_dump(exclude_none=True))
@router.patch('/{id}')
async def update_type(id: UUID, body: TypeBaseSchema):
return await ctrl.update(id, **body.model_dump(exclude_none=True))
@router.delete('/{id}')
async def delete_type(id: UUID):
return await ctrl.delete(id)
from uuid import UUID
from pydantic import BaseModel, ConfigDict
class TypeBaseSchema(BaseModel):
name: str | None = None
class TypeIdSchema(BaseModel):
id: UUID
model_config = ConfigDict(from_attributes=True)
class TypeSchema(TypeBaseSchema, TypeIdSchema):
pass
from uuid import UUID
from db.repositories import TypeRepository
from .schemas import TypeSchema
class TypeService:
type_repository = TypeRepository()
async def get_type_list(self, **kwargs):
types = await self.type_repository.get_list(**kwargs)
return [TypeSchema.model_validate(type) for type in types]
async def get_type(self, id: UUID):
type = await self.type_repository.get(id)
return TypeSchema.model_validate(type)
async def create_type(self, **kwargs):
type = await self.type_repository.create(**kwargs)
return TypeSchema.model_validate(type)
async def update_type(self, id: UUID, **kwargs):
type = await self.type_repository.update(id, **kwargs)
return TypeSchema.model_validate(type)
async def delete_type(self, id: UUID):
return await self.type_repository.delete(id)
import tortoise
from .models import (
Status, Type, Task, User, Comment
)
from exceptions import common as common_exc
......@@ -42,13 +45,21 @@ class BaseRepository:
raise common_exc.DeleteException(str(e))
class DirectorRepository(BaseRepository):
model = movie.Director
class StatusRepository(BaseRepository):
model = Status
class GenreRepository(BaseRepository):
model = movie.Genre
class TypeRepository(BaseRepository):
model = Type
class MovieRepository(BaseRepository):
model = movie.Movie
class TaskRepository(BaseRepository):
model = Task
class UserRepository(BaseRepository):
model = User
class CommentRepository(BaseRepository):
model = Comment
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment