backend.im/taskmanagerapi-i737xp
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Update code via agent code generation

Browse Source
This commit is contained in:
Automated Action 2025-06-12 18:28:09 +00:00
parent e2134243ae
commit bd40dcf387
25 changed files with 108 additions and 152 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

16
app/api/v1/endpoints/tasks.py
View File

@ -21,9 +21,7 @@ async def read_tasks(
""" """
Retrieve tasks. Retrieve tasks.
""" """
tasks = crud.task.get_multi_by_owner( tasks = crud.task.get_multi_by_owner(db=db, owner_id=current_user.id, skip=skip, limit=limit)
db=db, owner_id=current_user.id, skip=skip, limit=limit
)
return tasks return tasks
@ -53,9 +51,7 @@ async def read_task(
""" """
task = crud.task.get_by_id_and_owner(db=db, task_id=task_id, owner_id=current_user.id) task = crud.task.get_by_id_and_owner(db=db, task_id=task_id, owner_id=current_user.id)
if not task: if not task:
raise HTTPException( raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Task not found")
status_code=status.HTTP_404_NOT_FOUND, detail="Task not found"
)
return task return task
@ -72,9 +68,7 @@ async def update_task(
""" """
task = crud.task.get_by_id_and_owner(db=db, task_id=task_id, owner_id=current_user.id) task = crud.task.get_by_id_and_owner(db=db, task_id=task_id, owner_id=current_user.id)
if not task: if not task:
raise HTTPException( raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Task not found")
status_code=status.HTTP_404_NOT_FOUND, detail="Task not found"
)
task = crud.task.update(db=db, db_obj=task, obj_in=task_in) task = crud.task.update(db=db, db_obj=task, obj_in=task_in)
return task return task
@ -91,8 +85,6 @@ async def delete_task(
""" """
task = crud.task.get_by_id_and_owner(db=db, task_id=task_id, owner_id=current_user.id) task = crud.task.get_by_id_and_owner(db=db, task_id=task_id, owner_id=current_user.id)
if not task: if not task:
raise HTTPException( raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Task not found")
status_code=status.HTTP_404_NOT_FOUND, detail="Task not found"
)
task = crud.task.remove(db=db, id=task_id) task = crud.task.remove(db=db, id=task_id)
return task return task

4
app/core/config/settings.py
View File

@ -16,9 +16,7 @@ class Settings(BaseSettings):
ACCESS_TOKEN_EXPIRE_MINUTES: int = int(os.environ.get("ACCESS_TOKEN_EXPIRE_MINUTES", 60 * 24 * 8)) ACCESS_TOKEN_EXPIRE_MINUTES: int = int(os.environ.get("ACCESS_TOKEN_EXPIRE_MINUTES", 60 * 24 * 8))
# Database # Database
SQLALCHEMY_DATABASE_URL: str = os.environ.get( SQLALCHEMY_DATABASE_URL: str = os.environ.get("DATABASE_URL", "sqlite:////app/storage/db/db.sqlite")
"DATABASE_URL", "sqlite:////app/storage/db/db.sqlite"
)
class Config: class Config:
case_sensitive = True case_sensitive = True

8
app/core/security/__init__.py
View File

@ -6,10 +6,4 @@ from .security import (
verify_password, verify_password,
) )
__all__ = [ __all__ = ["create_access_token", "verify_password", "get_password_hash", "get_current_user", "get_current_active_user"]
"create_access_token",
"verify_password",
"get_password_hash",
"get_current_user",
"get_current_active_user"
]

17
app/core/security/security.py
View File

@ -22,15 +22,11 @@ pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
oauth2_scheme = OAuth2PasswordBearer(tokenUrl=f"{settings.API_V1_STR}/auth/login") oauth2_scheme = OAuth2PasswordBearer(tokenUrl=f"{settings.API_V1_STR}/auth/login")
def create_access_token( def create_access_token(subject: Union[str, Any], expires_delta: timedelta = None) -> str:
subject: Union[str, Any], expires_delta: timedelta = None
) -> str:
if expires_delta: if expires_delta:
expire = datetime.utcnow() + expires_delta expire = datetime.utcnow() + expires_delta
else: else:
expire = datetime.utcnow() + timedelta( expire = datetime.utcnow() + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
minutes=ACCESS_TOKEN_EXPIRE_MINUTES
)
to_encode = {"exp": expire, "sub": str(subject)} to_encode = {"exp": expire, "sub": str(subject)}
encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM) encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
return encoded_jwt return encoded_jwt
@ -44,18 +40,14 @@ def get_password_hash(password: str) -> str:
return pwd_context.hash(password) return pwd_context.hash(password)
async def get_current_user( async def get_current_user(db: Session = Depends(get_db), token: str = Depends(oauth2_scheme)) -> User:
db: Session = Depends(get_db), token: str = Depends(oauth2_scheme)
) -> User:
credentials_exception = HTTPException( credentials_exception = HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED, status_code=status.HTTP_401_UNAUTHORIZED,
detail="Could not validate credentials", detail="Could not validate credentials",
headers={"WWW-Authenticate": "Bearer"}, headers={"WWW-Authenticate": "Bearer"},
) )
try: try:
payload = jwt.decode( payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
token, SECRET_KEY, algorithms=[ALGORITHM]
)
token_data = TokenPayload(**payload) token_data = TokenPayload(**payload)
if token_data.sub is None: if token_data.sub is None:
raise credentials_exception raise credentials_exception
@ -63,6 +55,7 @@ async def get_current_user(
raise credentials_exception from e raise credentials_exception from e
from app.crud import user as user_crud from app.crud import user as user_crud
user = user_crud.get(db, id=token_data.sub) user = user_crud.get(db, id=token_data.sub)
if user is None: if user is None:
raise credentials_exception raise credentials_exception

12
app/crud/base.py
View File

@ -26,9 +26,7 @@ class CRUDBase(Generic[ModelType, CreateSchemaType, UpdateSchemaType]):
def get(self, db: Session, id: Any) -> Optional[ModelType]: def get(self, db: Session, id: Any) -> Optional[ModelType]:
return db.query(self.model).filter(self.model.id == id).first() return db.query(self.model).filter(self.model.id == id).first()
def get_multi( def get_multi(self, db: Session, *, skip: int = 0, limit: int = 100) -> List[ModelType]:
self, db: Session, *, skip: int = 0, limit: int = 100
) -> List[ModelType]:
return db.query(self.model).offset(skip).limit(limit).all() return db.query(self.model).offset(skip).limit(limit).all()
def create(self, db: Session, *, obj_in: CreateSchemaType) -> ModelType: def create(self, db: Session, *, obj_in: CreateSchemaType) -> ModelType:
@ -39,13 +37,7 @@ class CRUDBase(Generic[ModelType, CreateSchemaType, UpdateSchemaType]):
db.refresh(db_obj) db.refresh(db_obj)
return db_obj return db_obj
def update( def update(self, db: Session, *, db_obj: ModelType, obj_in: Union[UpdateSchemaType, Dict[str, Any]]) -> ModelType:
self,
db: Session,
*,
db_obj: ModelType,
obj_in: Union[UpdateSchemaType, Dict[str, Any]]
) -> ModelType:
obj_data = jsonable_encoder(db_obj) obj_data = jsonable_encoder(db_obj)
if isinstance(obj_in, dict): if isinstance(obj_in, dict):
update_data = obj_in update_data = obj_in

26
app/crud/crud_task.py
View File

@ -8,20 +8,10 @@ from app.schemas.task import TaskCreate, TaskUpdate
class CRUDTask(CRUDBase[Task, TaskCreate, TaskUpdate]): class CRUDTask(CRUDBase[Task, TaskCreate, TaskUpdate]):
def get_multi_by_owner( def get_multi_by_owner(self, db: Session, *, owner_id: int, skip: int = 0, limit: int = 100) -> List[Task]:
self, db: Session, *, owner_id: int, skip: int = 0, limit: int = 100 return db.query(self.model).filter(Task.owner_id == owner_id).offset(skip).limit(limit).all()
) -> List[Task]:
return (
db.query(self.model)
.filter(Task.owner_id == owner_id)
.offset(skip)
.limit(limit)
.all()
)
def create_with_owner( def create_with_owner(self, db: Session, *, obj_in: TaskCreate, owner_id: int) -> Task:
self, db: Session, *, obj_in: TaskCreate, owner_id: int
) -> Task:
obj_in_data = obj_in.model_dump() obj_in_data = obj_in.model_dump()
db_obj = self.model(**obj_in_data, owner_id=owner_id) db_obj = self.model(**obj_in_data, owner_id=owner_id)
db.add(db_obj) db.add(db_obj)
@ -29,14 +19,8 @@ class CRUDTask(CRUDBase[Task, TaskCreate, TaskUpdate]):
db.refresh(db_obj) db.refresh(db_obj)
return db_obj return db_obj
def get_by_id_and_owner( def get_by_id_and_owner(self, db: Session, *, task_id: int, owner_id: int) -> Optional[Task]:
self, db: Session, *, task_id: int, owner_id: int return db.query(self.model).filter(Task.id == task_id, Task.owner_id == owner_id).first()
) -> Optional[Task]:
return (
db.query(self.model)
.filter(Task.id == task_id, Task.owner_id == owner_id)
.first()
)
task = CRUDTask(Task) task = CRUDTask(Task)

4
app/crud/crud_user.py
View File

@ -28,9 +28,7 @@ class CRUDUser(CRUDBase[User, UserCreate, UserUpdate]):
db.refresh(db_obj) db.refresh(db_obj)
return db_obj return db_obj
def update( def update(self, db: Session, *, db_obj: User, obj_in: Union[UserUpdate, Dict[str, Any]]) -> User:
self, db: Session, *, db_obj: User, obj_in: Union[UserUpdate, Dict[str, Any]]
) -> User:
if isinstance(obj_in, dict): if isinstance(obj_in, dict):
update_data = obj_in update_data = obj_in
else: else:

5
app/db/session.py
View File

@ -11,10 +11,7 @@ DB_DIR.mkdir(parents=True, exist_ok=True)
SQLALCHEMY_DATABASE_URL = settings.SQLALCHEMY_DATABASE_URL SQLALCHEMY_DATABASE_URL = settings.SQLALCHEMY_DATABASE_URL
engine = create_engine( engine = create_engine(SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False})
SQLALCHEMY_DATABASE_URL,
connect_args={"check_same_thread": False}
)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine) SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

13
app/schemas/__init__.py
View File

@ -3,7 +3,14 @@ from .token import Token, TokenPayload
from .user import User, UserCreate, UserInDB, UserUpdate from .user import User, UserCreate, UserInDB, UserUpdate
__all__ = [ __all__ = [
"User", "UserCreate", "UserUpdate", "UserInDB", "User",
"Task", "TaskCreate", "TaskUpdate", "TaskInDB", "UserCreate",
"Token", "TokenPayload" "UserUpdate",
"UserInDB",
"Task",
"TaskCreate",
"TaskUpdate",
"TaskInDB",
"Token",
"TokenPayload",
] ]

10
main.py
View File

@ -1,4 +1,3 @@
import uvicorn import uvicorn
from fastapi import FastAPI from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware from fastapi.middleware.cors import CORSMiddleware
@ -27,19 +26,18 @@ app.add_middleware(
# Include API router # Include API router
app.include_router(api_router, prefix=settings.API_V1_STR) app.include_router(api_router, prefix=settings.API_V1_STR)
@app.get("/") @app.get("/")
async def root(): async def root():
"""Root endpoint returning service information""" """Root endpoint returning service information"""
return { return {"title": settings.PROJECT_NAME, "docs": "/docs", "health": "/health"}
"title": settings.PROJECT_NAME,
"docs": "/docs",
"health": "/health"
}
@app.get("/health", status_code=200) @app.get("/health", status_code=200)
async def health_check(): async def health_check():
"""Health check endpoint""" """Health check endpoint"""
return {"status": "healthy"} return {"status": "healthy"}
if __name__ == "__main__": if __name__ == "__main__":
uvicorn.run("main:app", host="0.0.0.0", port=8000, reload=True) uvicorn.run("main:app", host="0.0.0.0", port=8000, reload=True)

1
migrations/env.py
View File

@ -1,4 +1,3 @@
import os import os
import sys import sys
from logging.config import fileConfig from logging.config import fileConfig

74
migrations/versions/init_db.py
View File

@ -5,11 +5,12 @@ Revises:
Create Date: 2023-11-14 12:00:00.000000 Create Date: 2023-11-14 12:00:00.000000
""" """
import sqlalchemy as sa import sqlalchemy as sa
from alembic import op from alembic import op
# revision identifiers, used by Alembic. # revision identifiers, used by Alembic.
revision = '001_init_db' revision = "001_init_db"
down_revision = None down_revision = None
branch_labels = None branch_labels = None
depends_on = None depends_on = None
@ -18,51 +19,54 @@ depends_on = None
def upgrade(): def upgrade():
# Create users table # Create users table
op.create_table( op.create_table(
'user', "user",
sa.Column('id', sa.Integer(), nullable=False), sa.Column("id", sa.Integer(), nullable=False),
sa.Column('created_at', sa.DateTime(), nullable=True), sa.Column("created_at", sa.DateTime(), nullable=True),
sa.Column('updated_at', sa.DateTime(), nullable=True), sa.Column("updated_at", sa.DateTime(), nullable=True),
sa.Column('email', sa.String(), nullable=False), sa.Column("email", sa.String(), nullable=False),
sa.Column('username', sa.String(), nullable=False), sa.Column("username", sa.String(), nullable=False),
sa.Column('hashed_password', sa.String(), nullable=False), sa.Column("hashed_password", sa.String(), nullable=False),
sa.Column('is_active', sa.Boolean(), nullable=True), sa.Column("is_active", sa.Boolean(), nullable=True),
sa.Column('is_superuser', sa.Boolean(), nullable=True), sa.Column("is_superuser", sa.Boolean(), nullable=True),
sa.PrimaryKeyConstraint('id') sa.PrimaryKeyConstraint("id"),
) )
op.create_index(op.f('ix_user_email'), 'user', ['email'], unique=True) op.create_index(op.f("ix_user_email"), "user", ["email"], unique=True)
op.create_index(op.f('ix_user_id'), 'user', ['id'], unique=False) op.create_index(op.f("ix_user_id"), "user", ["id"], unique=False)
op.create_index(op.f('ix_user_username'), 'user', ['username'], unique=True) op.create_index(op.f("ix_user_username"), "user", ["username"], unique=True)
# Create tasks table # Create tasks table
op.create_table( op.create_table(
'task', "task",
sa.Column('id', sa.Integer(), nullable=False), sa.Column("id", sa.Integer(), nullable=False),
sa.Column('created_at', sa.DateTime(), nullable=True), sa.Column("created_at", sa.DateTime(), nullable=True),
sa.Column('updated_at', sa.DateTime(), nullable=True), sa.Column("updated_at", sa.DateTime(), nullable=True),
sa.Column('title', sa.String(100), nullable=False), sa.Column("title", sa.String(100), nullable=False),
sa.Column('description', sa.Text(), nullable=True), sa.Column("description", sa.Text(), nullable=True),
sa.Column('status', sa.Enum('todo', 'in_progress', 'done', name='taskstatus'), nullable=False), sa.Column("status", sa.Enum("todo", "in_progress", "done", name="taskstatus"), nullable=False),
sa.Column('priority', sa.Enum('low', 'medium', 'high', name='taskpriority'), nullable=False), sa.Column("priority", sa.Enum("low", "medium", "high", name="taskpriority"), nullable=False),
sa.Column('is_completed', sa.Boolean(), nullable=True), sa.Column("is_completed", sa.Boolean(), nullable=True),
sa.Column('owner_id', sa.Integer(), nullable=False), sa.Column("owner_id", sa.Integer(), nullable=False),
sa.ForeignKeyConstraint(['owner_id'], ['user.id'], ), sa.ForeignKeyConstraint(
sa.PrimaryKeyConstraint('id') ["owner_id"],
["user.id"],
),
sa.PrimaryKeyConstraint("id"),
) )
op.create_index(op.f('ix_task_id'), 'task', ['id'], unique=False) op.create_index(op.f("ix_task_id"), "task", ["id"], unique=False)
op.create_index(op.f('ix_task_title'), 'task', ['title'], unique=False) op.create_index(op.f("ix_task_title"), "task", ["title"], unique=False)
def downgrade(): def downgrade():
# Drop tasks table # Drop tasks table
op.drop_index(op.f('ix_task_title'), table_name='task') op.drop_index(op.f("ix_task_title"), table_name="task")
op.drop_index(op.f('ix_task_id'), table_name='task') op.drop_index(op.f("ix_task_id"), table_name="task")
op.drop_table('task') op.drop_table("task")
# Drop enum types (only works in PostgreSQL, not SQLite) # Drop enum types (only works in PostgreSQL, not SQLite)
# No need to drop enums in SQLite # No need to drop enums in SQLite
# Drop users table # Drop users table
op.drop_index(op.f('ix_user_username'), table_name='user') op.drop_index(op.f("ix_user_username"), table_name="user")
op.drop_index(op.f('ix_user_id'), table_name='user') op.drop_index(op.f("ix_user_id"), table_name="user")
op.drop_index(op.f('ix_user_email'), table_name='user') op.drop_index(op.f("ix_user_email"), table_name="user")
op.drop_table('user') op.drop_table("user")