contact working

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
+# from routers.user import get_current_active_user
-
-
-ACTIVE_USER = Annotated[alchemy.User, Depends(get_current_active_user)]
 
 __all__ = [
     'engine',
-    'ACTIVE_USER',
     'get_session',
 ]
 

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)
 

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

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')

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)

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)]

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