Test refine

alembic/versions/744577be90bf_add_created_at_and_updated_at_to_users.py

@@ -0,0 +1,34 @@
+"""add created_at and updated_at to users
+
+Revision ID: 744577be90bf
+Revises: dd5df88ee3df
+Create Date: 2026-02-24 22:16:44.362026
+
+"""
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+
+
+# revision identifiers, used by Alembic.
+revision: str = '744577be90bf'
+down_revision: Union[str, Sequence[str], None] = 'dd5df88ee3df'
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    """Upgrade schema."""
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.add_column('user', sa.Column('created_at', sa.DateTime(timezone=True), server_default=sa.text('now()'), nullable=False))
+    op.add_column('user', sa.Column('updated_at', sa.DateTime(timezone=True), server_default=sa.text('now()'), nullable=False))
+    # ### end Alembic commands ###
+
+
+def downgrade() -> None:
+    """Downgrade schema."""
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.drop_column('user', 'updated_at')
+    op.drop_column('user', 'created_at')
+    # ### end Alembic commands ###

api/deps.py

@@ -1,9 +1,10 @@
 from sqlmodel import Session
-from typing import Annotated
+from typing import Annotated, Any
 from fastapi import Depends, HTTPException
 from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
 from collections.abc import Generator
 
+from models import User
 from core.db import engine
 from core.security import local_token
 
@@ -28,3 +29,23 @@ def require_fixed_token(
 
 TokenDeps = Annotated[HTTPAuthorizationCredentials, Depends(require_fixed_token)]
 SessionDeps = Annotated[Session, Depends(get_db)]
+
+
+def parse_filters(params: dict[str, Any]) -> dict[str, tuple[str, Any]]:
+    filters = {}
+    for key, value in params.items():
+        if value is None or value == "":
+            continue
+
+        for operator in ["_ne", "_lte", "_lt", "_gte", "_gt", "_like"]:
+            if key.endswith(operator):
+                field_name = key[: -len(operator)]
+                filters[field_name] = (operator[1:], value)
+                break
+
+        else:
+            if key not in ["_start", "_end", "page", "_per_page", "_sort", "_order", "id"]:
+                if hasattr(User, key):
+                    filters[key] = ("eq", value)
+
+    return filters

api/routes/users.py

@@ -1,40 +1,81 @@
 import uuid
 import crud
 
-from crud import UserNotFound, UserAlreadyExists
-from fastapi import APIRouter, HTTPException
+from fastapi import APIRouter, HTTPException, Query, Request
+from typing import Literal, Annotated
 
-from api.deps import SessionDeps
+from api.deps import SessionDeps, parse_filters
+from crud import UserNotFound, UserAlreadyExists
 from models import (
     UserPublic,
     UserCreate,
+    UserCreateResponse,
     UserRegister,
     UserUpdate,
+    UserUpdateResponse,
     UsersPublic,
-    Message
+    Message,
 )
 
 
 router = APIRouter(prefix="/users", tags=["users"])
 
 
 @router.get("/", response_model=UsersPublic)
-def read_users(*, session: SessionDeps, skip: int = 0, limit: int = 100) -> UsersPublic:
-    users = crud.get_users(session=session, skip=skip, limit=limit)
-    return users
-
-
-@router.post("/", response_model=UserPublic)
-def create_user(*, user_in: UserCreate, session: SessionDeps) -> UserPublic:
+def read_users(
+        *,
+        session: SessionDeps,
+        request: Request,
+        _start: Annotated[int | None , Query(ge=0)] = None,
+        _end: Annotated[int | None, Query(ge=1)] = None,
+        page: Annotated[int | None, Query(ge=1)] = None,
+        _per_page: Annotated[int | None, Query(ge=1, le=100)] = None,
+        _sort: Annotated[str | None, Query()] = None,
+        _order: Annotated[Literal["asc", "desc"], Query()] = "asc",
+        id: Annotated[str | None, Query()] = None,
+) -> UsersPublic:
+    all_params = dict(request.query_params)
+
+    for key in ["_start", "_end", "page", "_per_page", "_sort", "_order", "id"]:
+        all_params.pop(key, None)
+
+    if id:
+        user_ids = [uuid.UUID(uid.strip()) for uid in id.split(",")]
+        users = crud.get_user_by_ids(session=session, user_ids=user_ids)
+        return UsersPublic(data=users, total=len(users))
+
+    if _start is not  None and _end is not None:
+        skip = _start
+        limit = _end - _start
+    elif page is not None and _per_page is not None:
+        skip = (page - 1) * _per_page
+        limit = _per_page
+
+    filters = parse_filters(all_params)
+
+    users = crud.get_users(
+        session=session,
+        skip=skip,
+        limit=limit,
+        sort_field=_sort,
+        sort_order=_order,
+        filters=filters,
+    )
+
+    return UsersPublic(data=users.data, total=users.total)
+
+
+@router.post("/", response_model=UserCreateResponse)
+def create_user(*, user_in: UserCreate, session: SessionDeps) -> UserCreateResponse:
     user = crud.get_user_by_email(email=user_in.email, session=session)
     if user:
         raise HTTPException(status_code=400, detail="Email already registered")
     user = crud.create_user(user_create=user_in, session=session)
     return user
 
 
-@router.post("/signup", response_model=UserPublic)
-def register_user(*, session: SessionDeps, user_in: UserRegister) -> UserPublic:
+@router.post("/signup", response_model=UserCreateResponse)
+def register_user(*, session: SessionDeps, user_in: UserRegister) -> UserCreateResponse:
     user = crud.get_user_by_email(email=user_in.email, session=session)
     if user:
         raise HTTPException(status_code=400, detail="Email already registered")
@@ -49,21 +90,21 @@ def read_user_by_id(*, session: SessionDeps, user_id: uuid.UUID) -> UserPublic |
     return user
 
 
-@router.patch("/{user_id}",response_model=UserPublic)
-def update_user(*, session: SessionDeps, user_id: uuid.UUID, user_in: UserUpdate) -> UserPublic:
+@router.patch("/{user_id}",response_model=UserUpdateResponse)
+def update_user(*, session: SessionDeps, user_id: uuid.UUID, user_in: UserUpdate) -> UserUpdateResponse:
     try:
         user = crud.update_user(session=session, user_id=user_id, user_in=user_in)
+        return user
     except UserNotFound:
         raise HTTPException(status_code=404, detail="User not found")
     except UserAlreadyExists:
         raise HTTPException(status_code=409, detail="User with this email already exists")
-    return user
 
 
 @router.delete("/{user_id}", response_model=Message)
 def delete_user(*, session: SessionDeps, user_id: uuid.UUID) -> Message:
     try:
-        crud.delete_user(session=session, user_id=user_id)
+        result = crud.delete_user(session=session, user_id=user_id)
+        return result
     except UserNotFound:
         raise HTTPException(status_code=404, detail="User not found")
-    return Message(message="User has been deleted!")

crud.py

@@ -2,8 +2,9 @@
 
 from pydantic import EmailStr
 from sqlmodel import Session, select, func
+from typing import Any, Literal
 
-from models import UserCreate, User, UserUpdate, Users
+from models import UserCreate, User, UserUpdate, Users, Message
 from core.security import get_password_hash
 
 
@@ -22,6 +23,12 @@ def get_user_by_id(*, user_id: uuid.UUID, session: Session) -> User | None:
     return db_user
 
 
+def get_user_by_ids(*, session: Session, user_ids: list[uuid.UUID]) -> list[User]:
+    statement = select(User).where(User.id.in_(user_ids))
+    users = session.exec(statement).all()
+    return users
+
+
 def get_user_by_email(*, email: EmailStr, session: Session) -> User | None:
     statement = select(User).where(User.email == email)
     db_user = session.exec(statement).first()
@@ -33,20 +40,54 @@ def get_users(
     session: Session,
     skip: int=0,
     limit: int=100,
+    sort_field: str | None =None,
+    sort_order: Literal["asc", "desc"]="asc",
+    filters: dict[str, tuple[str, Any]] | None=None,
     ) -> Users:
-    statement_count = select(func.count()).select_from(User)
-    count = session.exec(statement_count).one()
-    statement = select(User).offset(skip).limit(limit)
+    statement = select(User)
+    if filters:
+        for field_name, (operator, value) in filters.items():
+            if not hasattr(User, field_name):
+                continue
+            column = getattr(User, field_name)
+
+            if operator == "eq":
+                statement = statement.where(column == value)
+            elif operator == "ne":
+                statement = statement.where(column != value)
+            elif operator == "lt":
+                statement = statement.where(column < value)
+            elif operator == "lte":
+                statement = statement.where(column <= value)
+            elif operator == "gt":
+                statement = statement.where(column > value)
+            elif operator == "gte":
+                statement = statement.where(column >= value)
+            elif operator == "like":
+                statement = statement.where(column.ilike(f"%{value}%"))
+
+    count_statement = select(func.count()).select_from(statement)
+    count = session.exec(count_statement).one()
+
+    if sort_field and hasattr(User, sort_field):
+        column = getattr(User, sort_field)
+        if sort_order == "asc":
+            statement = statement.order_by(column.asc())
+        else:
+            statement = statement.order_by(column.desc())
+
+    statement = statement.offset(skip).limit(limit)
     users = session.exec(statement).all()
-    return Users(data=users, count=count)
+    return Users(data=users, total=count)
 
 
-def delete_user(*, user_id: uuid.UUID, session: Session) -> None:
+def delete_user(*, user_id: uuid.UUID, session: Session) -> Message:
     user_db = session.get(User, user_id)
     if not user_db:
         raise UserNotFound(user_id)
     session.delete(user_db)
     session.commit()
+    return Message(id=user_id)
 
 
 def create_user(*, user_create: UserCreate, session: Session) -> User:

models.py

@@ -1,6 +1,8 @@
 import uuid
 
-from sqlmodel import SQLModel, Field
+from datetime import datetime, timezone
+from sqlmodel import SQLModel, Field, Column
+from sqlalchemy import DateTime, func
 from pydantic import EmailStr
 
 
@@ -15,10 +17,29 @@ class User(UserBase, table=True):
     id: uuid.UUID = Field(default_factory=uuid.uuid4, primary_key=True)
     hashed_password: str
 
+    created_at: datetime = Field(
+        default_factory=lambda: datetime.now(timezone.utc),
+        sa_column=Column(
+            DateTime(timezone=True),
+            nullable=False,
+            server_default=func.now(),
+        ),
+    )
+
+    updated_at: datetime = Field(
+        default_factory=lambda: datetime.now(timezone.utc),
+        sa_column=Column(
+            DateTime(timezone=True),
+            nullable=False,
+            server_default=func.now(),
+            onupdate=func.now(),
+        ),
+    )
+
 
 class Users(SQLModel):
     data: list[User]
-    count: int
+    total: int
 
 
 class UserCreate(UserBase):
@@ -50,13 +71,23 @@ class UserPublic(UserBase):
     id: uuid.UUID
 
 
+class UserCreateResponse(UserBase):
+    id: uuid.UUID
+    created_at: datetime = Field(serialization_alias="createdAt")
+
+
+class UserUpdateResponse(UserBase):
+    id: uuid.UUID
+    updated_at: datetime = Field(serialization_alias="updatedAt")
+
+
 class UsersPublic(SQLModel):
     data: list[UserPublic]
-    count: int
+    total: int
 
 
 class Message(SQLModel):
-    message: str
+    id: uuid.UUID
 
 
 class BearerToken(SQLModel):