back to SQLmodel...

.gitignore

@@ -158,3 +158,4 @@ cython_debug/
 #  and can be added to the global gitignore or merged into this file.  For a more nuclear
 #  option (not recommended) you can uncomment the following to ignore the entire idea folder.
 #.idea/
+**/.DS_Store

ghostforge/db.py

@@ -1,22 +1,37 @@
 import os
 
 from fastapi import Depends
-from fastapi_users.db import SQLAlchemyBaseUserTableUUID
-from fastapi_users.db import SQLAlchemyUserDatabase
+from fastapi_users_db_sqlmodel import SQLModelBaseUserDB
+from fastapi_users_db_sqlmodel import SQLModelUserDatabaseAsync
 from sqlalchemy.ext.asyncio import AsyncSession
 from sqlalchemy.ext.asyncio import create_async_engine
-from sqlalchemy.orm import DeclarativeBase
 from sqlalchemy.orm import sessionmaker
 
+# from fastapi_users.db import SQLAlchemyBaseUserTableUUID
+# from fastapi_users.db import SQLAlchemyUserDatabase
+# from sqlalchemy.orm import DeclarativeBase
 
-class Base(DeclarativeBase):
+# from sqlmodel import SQLModel
+
+
+# class Base(DeclarativeBase):
+#     pass
+
+
+# Create a "Base" class for use with fastapi-users-db-sqlmodel and sqlalchemy
+class Base(SQLModelBaseUserDB):
     pass
 
 
-class User(SQLAlchemyBaseUserTableUUID, Base):
+# Create a "User" class for user with fastapi-users-db-sqlmodel
+class User(Base, table=True):
     pass
 
 
+# class User(SQLAlchemyBaseUserTableUUID, Base):
+#     pass
+
+
 database_url = (
     f'postgresql+asyncpg://{os.environ.get("POSTGRES_USER")}:'
     + f'{os.environ.get("POSTGRES_PASSWORD")}@'
@@ -41,4 +56,4 @@ async def create_db_and_tables():
 
 
 async def get_user_db(session: AsyncSession = Depends(get_session)):
-    yield SQLAlchemyUserDatabase(session, User)
+    yield SQLModelUserDatabaseAsync(session, User)

ghostforge/ghosts.py

@@ -0,0 +1,105 @@
+from datetime import datetime
+from datetime import timezone
+from typing import Annotated
+from typing import Optional
+
+from fastapi import APIRouter
+from fastapi import Depends
+from fastapi import HTTPException
+from fastapi import Query
+from fastapi import Request
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+from sqlalchemy.orm import Session
+from sqlmodel import Column
+from sqlmodel import Field
+from sqlmodel import SQLModel
+from sqlmodel import TIMESTAMP
+
+from ghostforge.db import get_session
+from ghostforge.db import User
+from ghostforge.htmljson import HtmlJson
+from ghostforge.users import get_current_user
+
+gf = APIRouter()
+hj = HtmlJson()
+
+
+class Ghost(SQLModel, table=True):
+    id: Optional[int] = Field(primary_key=True)
+    first_name: Optional[str] = Field()
+    last_name: Optional[str] = Field()
+    middle_name: Optional[str] = Field()
+    birthdate: Optional[datetime] = Field(
+        sa_column=Column(
+            TIMESTAMP(timezone=True),
+        )
+    )
+
+
+class GhostCreate(Ghost, table=False):
+    pass
+
+
+class GhostUpdate(Ghost, table=False):
+    pass
+
+
+class GhostRead(Ghost, table=False):
+    pass
+
+
+@gf.get("/ghosts/{ghost_id}")
+@hj.html_or_json("ghost.html")
+async def read_ghost(
+    ghost_id: int,
+    current_user: Annotated[User, Depends(get_current_user())],
+    session: AsyncSession = Depends(get_session),
+    request: Request = None,
+):
+    ghost = await session.get(Ghost, ghost_id)
+    data = {"crumbs": [("ghosts", "/ghosts"), (ghost.id, False)]}
+    request.state.ghostforge = data | getattr(request.state, "ghostforge", {})
+    return ghost
+
+
+@gf.put("/ghosts/{ghost_id}")
+async def update_ghost(
+    ghost_id: int,
+    ghost: GhostUpdate,
+    current_user: Annotated[User, Depends(get_current_user())],
+    session: AsyncSession = Depends(get_session),
+    request: Request = None,
+):
+    db_ghost = await session.get(Ghost, ghost_id)
+    if db_ghost is None:
+        raise HTTPException(status_code=404, detail="No ghost with that ID")
+    for k, v in ghost.dict().items():
+        setattr(db_ghost, k, v)
+    await session.commit()
+    await session.refresh(db_ghost)
+    return db_ghost
+
+
+@gf.post("/ghosts")
+async def create_ghost(
+    ghost: GhostCreate,
+    current_user: Annotated[User, Depends(get_current_user())],
+    session: AsyncSession = Depends(get_session),
+    request: Request = None,
+):
+    new_ghost = Ghost.from_orm(ghost)
+    ghost.birthdate = ghost.birthdate.replace(tzinfo=timezone.utc)
+    print(ghost.birthdate)
+    session.add(new_ghost)
+    await session.commit()
+    await session.refresh(new_ghost)
+    return new_ghost
+
+
+@gf.get("/ghosts")
+async def read_users(
+    offset: int = 0, limit: int = Query(default=100, lte=100), session: Session = Depends(get_session)
+):
+    ghosts = await session.execute(select(Ghost).offset(offset).limit(limit))
+    return ghosts.scalars().all()

ghostforge/serve.py

@@ -10,7 +10,7 @@ from fastapi.staticfiles import StaticFiles
 
 from ghostforge.db import create_db_and_tables
 from ghostforge.db import User
-from ghostforge.htmljson import HtmlJson
+from ghostforge.ghosts import gf as gf_ghosts
 from ghostforge.templates import templates
 from ghostforge.users import fastapi_users
 from ghostforge.users import get_current_user
@@ -20,15 +20,10 @@ from ghostforge.users import UserRead
 from ghostforge.users import UserUpdate
 from ghostforge.users import web_backend
 
-# from ghostforge.users import gf as gf_users
-
-# from ghostforge.models import User
 
 gf = FastAPI()
-hj = HtmlJson()
 gf.mount("/static", StaticFiles(directory="ghostforge/static"), name="static")
-
-# gf.include_router(gf_users)
+gf.include_router(gf_ghosts)
 
 gf.include_router(fastapi_users.get_auth_router(jwt_backend), prefix="/auth/jwt", tags=["auth"])
 gf.include_router(fastapi_users.get_auth_router(web_backend), prefix="/auth/cookie", tags=["auth"])
@@ -54,6 +49,11 @@ gf.include_router(
 )
 
 
+@gf.on_event("startup")
+async def on_startup():
+    await create_db_and_tables()
+
+
 @gf.get("/", response_class=HTMLResponse)
 async def home(request: Request, current_user=Depends(get_current_user())):
     if current_user:
@@ -77,15 +77,8 @@ async def logout(response: Response):
 
 
 @gf.get("/dashboard")
-async def authenticated_route(
-    request: Request, current_user: Annotated[User, Depends(get_current_user(optional=True))]
-):
+async def dashboard(request: Request, current_user: Annotated[User, Depends(get_current_user(optional=True))]):
     if not current_user:
         return RedirectResponse(url="/login")
-    crumbs = [("dashboard", False), (current_user.email, False)]
+    crumbs = [("dashboard", False)]
     return templates.TemplateResponse("dashboard.html", {"request": request, "crumbs": crumbs, "user": current_user})
-
-
-@gf.on_event("startup")
-async def on_startup():
-    await create_db_and_tables()

ghostforge/templates/ghost.html

@@ -0,0 +1,16 @@
+{% extends "base.html" %}
+
+{% block content %}
+<div class="columns">
+    <div class="column col-12">
+        <h1>Words!</h1>
+    </div>
+</div>
+<div class="columns">
+    <div class="column col-3"></div>
+    <div class="column col-6">
+        <h5>More words.</h5>
+    </div>
+    <div class="column col-3"></div>
+</div>
+{% endblock content %}

ghostforge/templates/navigation/top.html

@@ -2,6 +2,7 @@
     <section class="navbar-section">
         {% if crumbs %}
         <ul class="breadcrumb">
+            <li class="breadcrumb-item"><kbd>{{ user.email.split('@')[0] }}</kbd></li>
             {% for name, url in crumbs %}
             <li class="breadcrumb-item">
                 {% if url %}

ghostforge/users.py

@@ -13,11 +13,12 @@ from fastapi_users.authentication import AuthenticationBackend
 from fastapi_users.authentication import BearerTransport
 from fastapi_users.authentication import CookieTransport
 from fastapi_users.authentication import JWTStrategy
-from fastapi_users.db import SQLAlchemyUserDatabase
+from fastapi_users_db_sqlmodel import SQLModelUserDatabase
 
 from ghostforge.db import get_user_db
 from ghostforge.db import User
 
+
 SECRET = os.environ.get("GHOSTFORGE_JWT_SECRET")
 
 gf = APIRouter()
@@ -49,7 +50,7 @@ class UserManager(UUIDIDMixin, BaseUserManager[User, uuid.UUID]):
         print(f"Verification requested for user {user.id}. Verification token: {token}")
 
 
-async def get_user_manager(user_db: SQLAlchemyUserDatabase = Depends(get_user_db)):
+async def get_user_manager(user_db: SQLModelUserDatabase = Depends(get_user_db)):
     yield UserManager(user_db)
 
 

pyproject.toml

@@ -21,8 +21,8 @@ dependencies = [
     "greenlet==2.0.2",
     "jinja2==3.1.2",
     "fastapi-users==11.0.0",
-    "fastapi-users-db-sqlalchemy==5.0.0",
-    "httpx==0.24.1",
+    "fastapi-users-db-sqlmodel==0.3.0",
+    "sqlmodel==0.0.8",
 ]
 
 [project.urls]