350b445f45
- Setup project structure and FastAPI application - Create SQLite database with SQLAlchemy - Implement user authentication with JWT - Create task and user models - Add CRUD operations for tasks and users - Configure Alembic for database migrations - Implement API endpoints for task management - Add error handling and validation - Configure CORS middleware - Create health check endpoint - Add comprehensive documentation
35 lines
1.0 KiB
Python
35 lines
1.0 KiB
Python
from typing import List
|
|
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
from sqlalchemy.future import select
|
|
|
|
from app.core.security import generate_uuid
|
|
from app.crud.base import CRUDBase
|
|
from app.models.task import Task
|
|
from app.schemas.task import TaskCreate, TaskUpdate
|
|
|
|
|
|
class CRUDTask(CRUDBase[Task, TaskCreate, TaskUpdate]):
|
|
async def create_with_owner(
|
|
self, db: AsyncSession, *, obj_in: TaskCreate, owner_id: str
|
|
) -> Task:
|
|
obj_in_data = obj_in.dict()
|
|
db_obj = Task(**obj_in_data, id=generate_uuid(), user_id=owner_id)
|
|
db.add(db_obj)
|
|
await db.commit()
|
|
await db.refresh(db_obj)
|
|
return db_obj
|
|
|
|
async def get_multi_by_owner(
|
|
self, db: AsyncSession, *, owner_id: str, skip: int = 0, limit: int = 100
|
|
) -> List[Task]:
|
|
result = await db.execute(
|
|
select(Task)
|
|
.filter(Task.user_id == owner_id)
|
|
.offset(skip)
|
|
.limit(limit)
|
|
)
|
|
return result.scalars().all()
|
|
|
|
|
|
task = CRUDTask(Task) |