From 5614ecbba6e5db833f193e02e6ab809571d0277c Mon Sep 17 00:00:00 2001 From: "Bernhard Radermacher (hakisto)" Date: Sat, 30 Aug 2025 10:06:47 +0200 Subject: [PATCH] contact working --- app/dependencies/__init__.py | 6 +- app/main.py | 2 + app/routers/__init__.py | 1 + app/routers/contact.py | 117 +++++++++++++++++++++++++++++++++++ app/routers/status.py | 20 +++--- app/routers/user.py | 7 ++- app/utils/__init__.py | 50 +++++++++++++++ 7 files changed, 185 insertions(+), 18 deletions(-) create mode 100644 app/routers/contact.py create mode 100644 app/utils/__init__.py diff --git a/app/dependencies/__init__.py b/app/dependencies/__init__.py index 4b04fe0..6181ee3 100644 --- a/app/dependencies/__init__.py +++ b/app/dependencies/__init__.py @@ -7,14 +7,10 @@ from .engine import engine from .session import get_session -from routers.user import get_current_active_user - - -ACTIVE_USER = Annotated[alchemy.User, Depends(get_current_active_user)] +# from routers.user import get_current_active_user __all__ = [ 'engine', - 'ACTIVE_USER', 'get_session', ] diff --git a/app/main.py b/app/main.py index 65ccf95..88861e5 100644 --- a/app/main.py +++ b/app/main.py @@ -19,6 +19,8 @@ app = FastAPI(lifespan=lifespan) app.add_middleware(CORSMiddleware, allow_origins=["*"], allow_methods=["*"]) +app.include_router(routers.contact) + app.include_router(routers.status) app.include_router(routers.user) diff --git a/app/routers/__init__.py b/app/routers/__init__.py index 984f0ee..e676583 100644 --- a/app/routers/__init__.py +++ b/app/routers/__init__.py @@ -1,2 +1,3 @@ +from .contact import router as contact from .status import router as status from .user import router as user diff --git a/app/routers/contact.py b/app/routers/contact.py new file mode 100644 index 0000000..ad95a38 --- /dev/null +++ b/app/routers/contact.py @@ -0,0 +1,117 @@ +from typing import Annotated + +from fastapi import APIRouter, Query, Depends +from sqlmodel import SQLModel, Field +from sqlmodel import select + +import alchemy +import utils +from dependencies import get_session +from utils import update_item, create_item +from .status_model import Status +from .user import ACTIVE_USER + +PRIMARY_ANNOTATION = utils.make_primary_annotation('Contact') + + +# ---------------------------------------------------------------- +# Models + +class ContactBase(SQLModel): + name: str = Field(max_length=80, unique=True) + address: str = Field(max_length=253) + notes: str | None + + +class Contact(ContactBase): + id: int | None = Field(default=None, primary_key=True) + + +class ContactCreate(ContactBase): + address: str | None = None + notes: str | None = None + + +class ContactPublic(ContactBase): + id: int + status: Status + + +class ContactUpdate(ContactBase): + name: str | None = None + address: str | None = None + notes: str | None = None + + +# ---------------------------------------------------------------- +# Routes + +router = APIRouter(prefix="/contact", tags=["contact"]) + + +@router.get("/", response_model=list[ContactPublic]) +async def get_contacts( + offset: int = 0, + limit: Annotated[int, Query(le=100)] = 100, + session=Depends(get_session)): + """Get list of all contacts""" + return session.exec(select(alchemy.Contact).offset(offset).limit(limit)).all() + + +@router.get("/{contact_id}", + response_model=ContactPublic, + responses={404: {"description": "Not found"}}) +async def get_contact( + contact_id: PRIMARY_ANNOTATION, + session=Depends(get_session)): + result = utils.get_single_record(session, alchemy.Contact, contact_id) + return result + + +@router.post("/", response_model=ContactPublic) +async def create_contact( + contact: ContactCreate, + current_user: ACTIVE_USER, + session=Depends(get_session)): + return create_item( + session=session, + cls=alchemy.Contact, + current_user=current_user, + data=Contact.model_validate(contact)) + + +@router.patch("/{contact_id}", response_model=ContactPublic) +async def update_contact( + contact_id: PRIMARY_ANNOTATION, + contact: ContactUpdate, + current_user: ACTIVE_USER, + session=Depends(get_session)): + return update_item( + session=session, + current_user=current_user, + item=utils.get_single_record(session, alchemy.Contact, contact_id), + data=contact) + + +@router.put("/{contact_id}/activate", response_model=ContactPublic) +async def activate_contact( + contact_id: PRIMARY_ANNOTATION, + current_user: ACTIVE_USER, + session=Depends(get_session)): + return utils.set_item_status( + session=session, + current_user=current_user, + item=utils.get_single_record(session, alchemy.Contact, contact_id), + status='A') + + +@router.put("/{contact_id}/deactivate", response_model=ContactPublic) +async def deactivate_contact( + contact_id: PRIMARY_ANNOTATION, + current_user: ACTIVE_USER, + session=Depends(get_session)): + return utils.set_item_status( + session=session, + current_user=current_user, + item=utils.get_single_record(session, alchemy.Contact, contact_id), + status='I') diff --git a/app/routers/status.py b/app/routers/status.py index d440019..07d18e1 100644 --- a/app/routers/status.py +++ b/app/routers/status.py @@ -1,7 +1,7 @@ from fastapi import APIRouter, Depends, HTTPException from sqlmodel import select, Session -from dependencies import get_session, ACTIVE_USER +from dependencies import get_session from routers.status_model import Status router = APIRouter( @@ -9,15 +9,6 @@ router = APIRouter( tags=["status"], ) - -@router.get("/", response_model=list[Status]) -async def get_statuses( - current_user: ACTIVE_USER, - session=Depends(get_session)): - """Get list of all statuses""" - return session.exec(select(Status)).all() - - def _get_status(status: str, session: Session): result = session.get(Status, status) if result is None: @@ -26,9 +17,16 @@ def _get_status(status: str, session: Session): raise HTTPException(status_code=404, detail=f"Status {status!r} not found") return result + +@router.get("/", response_model=list[Status]) +async def get_statuses( + session=Depends(get_session)): + """Get list of all statuses""" + return session.exec(select(Status)).all() + + @router.get("/{status}", responses={404: {"description": "Not found"}}) async def get_status( status: str, - current_user: ACTIVE_USER, session=Depends(get_session)): return _get_status(status, session) diff --git a/app/routers/user.py b/app/routers/user.py index 6a81a8d..7a23301 100644 --- a/app/routers/user.py +++ b/app/routers/user.py @@ -10,11 +10,11 @@ from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm from jwt.exceptions import InvalidTokenError from passlib.context import CryptContext from pydantic import BaseModel -from sqlmodel import Session, SQLModel, Field, select +from sqlmodel import SQLModel, Field, Session, select import alchemy from dependencies import get_session -from .status_model import Status +from routers.status_model import Status logging.getLogger('passlib').setLevel(logging.ERROR) @@ -284,3 +284,6 @@ async def login_for_access_token( session=Depends(get_session), ) -> Token: return _process_login(form_data.username, form_data.password, session) + + +ACTIVE_USER = Annotated[alchemy.User, Depends(get_current_active_user)] diff --git a/app/utils/__init__.py b/app/utils/__init__.py new file mode 100644 index 0000000..93074f5 --- /dev/null +++ b/app/utils/__init__.py @@ -0,0 +1,50 @@ + +from typing import Annotated + +from fastapi import Path, HTTPException +from sqlalchemy import select + + +def make_primary_annotation(name=str): + return Annotated[str, Path(description=f'{name}, either id (int) or name')] + + +def get_single_record(session, cls, primary: str): + print(session) + print(cls) + print(primary) + result = session.get(cls, primary) + print(result) + if result is None: + result = session.scalar(select(cls).where(cls.name == primary)) + if result is None: + raise HTTPException( + status_code=404, + detail=f"{cls.__class__.__name__} {primary!r} not found.") + return result + + +def set_item_status(session, current_user, item, status: str): + item.status_id = status + item._user__id = current_user.id + session.commit() + session.refresh(item) + return item + + +def update_item(session, current_user, item, data): + for k, v in data.model_dump(exclude_unset=True).items(): + setattr(item, k, v) + item._user__id = current_user.id + session.commit() + session.refresh(item) + return item + + +def create_item(session, cls, current_user, data): + item = cls(**data.model_dump()) + item._user__id = current_user.id + session.add(item) + session.commit() + session.refresh(item) + return item