Preparing to setup fastapi-users

ghostforge/users.py

@@ -1,15 +1,109 @@
-from fastapi import APIRouter
-from fastapi import Request
-from fastapi.responses import HTMLResponse
+from datetime import datetime
+from typing import List
+from typing import Optional
 
-from templates import templates
+from argon2 import PasswordHasher
+from fastapi import APIRouter
+from fastapi import Depends
+from fastapi import HTTPException
+from fastapi import Query
+from fastapi import Request
+from sqlmodel import Field
+from sqlmodel import select
+from sqlmodel import Session
+from sqlmodel import SQLModel
+
+from ghostforge.db import get_session
+from ghostforge.htmljson import HtmlJson
 
 gf = APIRouter()
+ph = PasswordHasher()
+hj = HtmlJson()
 
 
-@gf.get("/user/{id}", response_class=HTMLResponse)
-async def get_user(request: Request, id: str):
-    crumbs = [("settings", False), ("users", "/users"), (id, False)]
-    return templates.TemplateResponse(
-        "user.html", {"request": request, "crumbs": crumbs, "user": "test", "id": id, "title": f"User {id}"}
-    )
+class UserBase(SQLModel):
+    name: str = Field()
+
+    class Config:
+        orm_mode = True
+
+
+class User(UserBase, table=True):
+    id: Optional[int] = Field(default=None, primary_key=True)
+    creation: datetime = Field(default=datetime.now())
+    password_hash: Optional[str]
+
+    def verify_password(self, password: str):
+        return ph.verify(self.password_hash, password)
+
+    def set_password(self, password: str):
+        self.password_hash = ph.hash(password)
+
+
+class UserCreate(UserBase):
+    name: str
+    password: str
+
+
+class UserRead(UserBase):
+    id: int
+
+
+class UserUpdate(SQLModel):
+    name: Optional[str] = None
+
+
+@gf.get("/users", response_model=List[UserRead])
+async def read_users(
+    offset: int = 0, limit: int = Query(default=100, lte=100), session: Session = Depends(get_session)
+):
+    users = await session.execute(select(User).offset(offset).limit(limit))
+    return [UserRead.from_orm(user[0]) for user in users.all()]
+
+
+@gf.post("/users", response_model=UserRead)
+async def create_hero(user: UserCreate, session: Session = Depends(get_session)):
+    new_user = User.from_orm(user)
+    if len(user.password) < 12:
+        raise HTTPException(status_code=442, detail="Password must be at least 12 characters")
+    new_user.set_password(user.password)
+    session.add(new_user)
+    await session.commit()
+    await session.refresh(new_user)
+    return new_user
+
+
+@gf.get("/users/{user_id}", response_model=UserRead)
+@hj.html_or_json("user.html")
+async def read_user(user_id: int, session: Session = Depends(get_session), request: Request = None):
+    user = await session.get(User, user_id)
+    if not user:
+        raise HTTPException(status_code=404, detail="User not found")
+
+    data = {"crumbs": [("settings", False), ("users", "/users"), (user_id, False)]}
+    request.state.ghostforge = data | getattr(request.state, "ghostforge", {})
+    return user
+
+
+@gf.patch("/users/{user_id}", response_model=UserRead)
+async def update_user(user_id: int, user: UserUpdate, session: Session = Depends(get_session), request: Request = None):
+    edit_user = await session.get(User, user_id)
+    if not edit_user:
+        raise HTTPException(status_code=404, detail="User not found")
+    data = user.dict(exclude_unset=True)
+    for key, value in data.items():
+        setattr(edit_user, key, value)
+    session.add(edit_user)
+    await session.commit()
+    await session.refresh(edit_user)
+    return edit_user
+
+
+@gf.delete("/users/{user_id}")
+async def delete_user(user_id: int, session: Session = Depends(get_session), request: Request = None):
+    user = await session.get(User, user_id)
+    if not user:
+        raise HTTPException(status_code=404, detail="User not found")
+    session.delete(user)
+    await session.commit()
+    return {"ok": True}