Automated deployment: Mon Feb 3 09:51:52 UTC 2020 b7bb7c2e31abd420ac3668443d2a57f798120e53

src/full_sqlalchemy.py

@@ -0,0 +1,74 @@
+import databases
+import sqlalchemy
+from fastapi import FastAPI
+from fastapi_users import FastAPIUsers, models
+from fastapi_users.authentication import JWTAuthentication
+from fastapi_users.db import SQLAlchemyBaseUserTable, SQLAlchemyUserDatabase
+from sqlalchemy.ext.declarative import DeclarativeMeta, declarative_base
+
+DATABASE_URL = "sqlite:///./test.db"
+SECRET = "SECRET"
+
+
+class User(models.BaseUser):
+    pass
+
+
+class UserCreate(User, models.BaseUserCreate):
+    pass
+
+
+class UserUpdate(User, models.BaseUserUpdate):
+    pass
+
+
+class UserDB(User, models.BaseUserDB):
+    pass
+
+
+database = databases.Database(DATABASE_URL)
+Base: DeclarativeMeta = declarative_base()
+
+
+class UserTable(Base, SQLAlchemyBaseUserTable):
+    pass
+
+
+engine = sqlalchemy.create_engine(
+    DATABASE_URL, connect_args={"check_same_thread": False}
+)
+Base.metadata.create_all(engine)
+
+users = UserTable.__table__
+user_db = SQLAlchemyUserDatabase(UserDB, database, users)
+
+
+auth_backends = [
+    JWTAuthentication(secret=SECRET, lifetime_seconds=3600),
+]
+
+app = FastAPI()
+fastapi_users = FastAPIUsers(
+    user_db, auth_backends, User, UserCreate, UserUpdate, UserDB, SECRET,
+)
+app.include_router(fastapi_users.router, prefix="/users", tags=["users"])
+
+
+@fastapi_users.on_after_register()
+def on_after_register(user: User):
+    print(f"User {user.id} has registered.")
+
+
+@fastapi_users.on_after_forgot_password()
+def on_after_forgot_password(user: User, token: str):
+    print(f"User {user.id} has forgot their password. Reset token: {token}")
+
+
+@app.on_event("startup")
+async def startup():
+    await database.connect()
+
+
+@app.on_event("shutdown")
+async def shutdown():
+    await database.disconnect()