Create a User model and API endpoints for creating and getting

Author: ArtyomVancyan

examples/demonstration/database.py

@@ -0,0 +1,21 @@
+from sqlalchemy import create_engine
+from sqlalchemy.ext.declarative import declarative_base
+from sqlalchemy.orm import sessionmaker
+
+engine = create_engine(
+    "sqlite:///./database.sqlite",
+    connect_args={
+        "check_same_thread": False,
+    },
+)
+
+Base = declarative_base()
+SessionLocal = sessionmaker(bind=engine, autoflush=False)
+
+
+def get_db():
+    db = SessionLocal()
+    try:
+        yield db
+    finally:
+        db.close()

examples/demonstration/main.py

@@ -2,10 +2,14 @@
 from fastapi import FastAPI
 
 from config import oauth2_config
+from database import Base
+from database import engine
 from fastapi_oauth2.middleware import OAuth2Middleware
 from fastapi_oauth2.router import router as oauth2_router
 from router import router as app_router
 
+Base.metadata.create_all(bind=engine)
+
 router = APIRouter()
 
 app = FastAPI()

examples/demonstration/models.py

@@ -0,0 +1,23 @@
+from sqlalchemy import Column
+from sqlalchemy import Integer
+from sqlalchemy import String
+from sqlalchemy.orm import Session
+
+from database import Base
+
+
+class BaseModel(Base):
+    __abstract__ = True
+
+    def save(self, db: Session):
+        db.add(self)
+        db.commit()
+        db.refresh(self)
+        return self
+
+
+class User(BaseModel):
+    __tablename__ = "users"
+
+    id = Column(Integer, primary_key=True, index=True)
+    username = Column(String)

examples/demonstration/router.py

@@ -1,12 +1,15 @@
 import json
 
+from fastapi import APIRouter
 from fastapi import Depends
 from fastapi import Request
-from fastapi import APIRouter
 from fastapi.responses import HTMLResponse
 from fastapi.templating import Jinja2Templates
+from sqlalchemy.orm import Session
 
+from database import get_db
 from fastapi_oauth2.security import OAuth2
+from models import User
 
 oauth2 = OAuth2()
 router = APIRouter()
@@ -19,5 +22,16 @@ async def root(request: Request):
 
 
 @router.get("/user")
-def user(request: Request, _: str = Depends(oauth2)):
+def user_get(request: Request, _: str = Depends(oauth2)):
     return request.user
+
+
+@router.get("/users")
+def users_get(request: Request, db: Session = Depends(get_db), _: str = Depends(oauth2)):
+    return db.query(User).all()
+
+
+@router.post("/users")
+async def users_post(request: Request, db: Session = Depends(get_db), _: str = Depends(oauth2)):
+    data = await request.json()
+    return User(**data).save(db)