python/fastapi-users
1
0
Fork 0
mirror of https://github.com/fastapi-users/fastapi-users.git synced 2026-03-13 07:49:55 +08:00
Code Issues Packages Projects Releases Wiki Activity

Rework the documentation

Browse Source
This commit is contained in:
François Voron
2021-09-17 15:14:36 +02:00
parent e31a0a99b5
commit fc59cf11ef
33 changed files with 486 additions and 861 deletions
Download Patch File Download Diff File Expand all files Collapse all files

25
docs/src/db_mongodb.py
View File

@@ -1,24 +1,7 @@
import motor.motor_asyncio
from fastapi import FastAPI
from fastapi_users import models
from fastapi_users.db import MongoDBUserDatabase
class User(models.BaseUser):
pass
class UserCreate(models.BaseUserCreate):
pass
class UserUpdate(User, models.BaseUserUpdate):
pass
class UserDB(User, models.BaseUserDB):
pass
from .models import UserDB
DATABASE_URL = "mongodb://localhost:27017"
client = motor.motor_asyncio.AsyncIOMotorClient(
@@ -28,7 +11,5 @@ db = client["database_name"]
collection = db["users"]
app = FastAPI()
user_db = MongoDBUserDatabase(UserDB, collection)
def get_user_db():
yield MongoDBUserDatabase(UserDB, collection)

37
docs/src/db_ormar.py
View File

@@ -1,26 +1,8 @@
import databases
import sqlalchemy
from fastapi import FastAPI
from fastapi_users import models
from fastapi_users.db import OrmarBaseUserModel, OrmarUserDatabase
class User(models.BaseUser):
pass
class UserCreate(models.BaseUserCreate):
pass
class UserUpdate(User, models.BaseUserUpdate):
pass
class UserDB(User, models.BaseUserDB):
pass
DATABASE_URL = "sqlite:///test.db"
metadata = sqlalchemy.MetaData()
database = databases.Database(DATABASE_URL)
@@ -37,20 +19,5 @@ engine = sqlalchemy.create_engine(DATABASE_URL)
metadata.create_all(engine)
user_db = OrmarUserDatabase(UserDB, UserModel)
app = FastAPI()
app.state.database = database
@app.on_event("startup")
async def startup() -> None:
database_ = app.state.database
if not database_.is_connected:
await database_.connect()
@app.on_event("shutdown")
async def shutdown() -> None:
database_ = app.state.database
if database_.is_connected:
await database_.disconnect()
def get_user_db():
yield OrmarUserDatabase(UserDB, UserModel)

35
docs/src/db_sqlalchemy.py
View File

@@ -1,31 +1,12 @@
import databases
import sqlalchemy
from fastapi import FastAPI
from fastapi_users import models
from fastapi_users.db import SQLAlchemyBaseUserTable, SQLAlchemyUserDatabase
from sqlalchemy.ext.declarative import DeclarativeMeta, declarative_base
class User(models.BaseUser):
pass
class UserCreate(models.BaseUserCreate):
pass
class UserUpdate(User, models.BaseUserUpdate):
pass
class UserDB(User, models.BaseUserDB):
pass
from .models import UserDB
DATABASE_URL = "sqlite:///./test.db"
database = databases.Database(DATABASE_URL)
Base: DeclarativeMeta = declarative_base()
@@ -36,20 +17,10 @@ class UserTable(Base, SQLAlchemyBaseUserTable):
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)
app = FastAPI()
@app.on_event("startup")
async def startup():
await database.connect()
@app.on_event("shutdown")
async def shutdown():
await database.disconnect()
def get_user_db():
yield SQLAlchemyUserDatabase(UserDB, database, users)

40
docs/src/db_tortoise.py
View File

@@ -1,40 +0,0 @@
from fastapi import FastAPI
from fastapi_users import models
from fastapi_users.db import TortoiseBaseUserModel, TortoiseUserDatabase
from tortoise.contrib.fastapi import register_tortoise
from tortoise.contrib.pydantic import PydanticModel
class UserModel(TortoiseBaseUserModel):
pass
class User(models.BaseUser):
pass
class UserCreate(models.BaseUserCreate):
pass
class UserUpdate(User, models.BaseUserUpdate):
pass
class UserDB(User, models.BaseUserDB, PydanticModel):
class Config:
orm_mode = True
orig_model = UserModel
DATABASE_URL = "sqlite://./test.db"
user_db = TortoiseUserDatabase(UserDB, UserModel)
app = FastAPI()
register_tortoise(
app,
db_url=DATABASE_URL,
modules={"models": ["path_to_your_package"]},
generate_schemas=True,
)

9
docs/src/db_tortoise_adapter.py Normal file
View File

@@ -0,0 +1,9 @@
from fastapi_users.db import TortoiseUserDatabase
from .models import UserDB, UserModel
DATABASE_URL = "sqlite://./test.db"
def get_user_db():
yield TortoiseUserDatabase(UserDB, UserModel)

25
docs/src/db_tortoise_model.py Normal file
View File

@@ -0,0 +1,25 @@
from fastapi_users import models
from fastapi_users.db import TortoiseBaseUserModel
from tortoise.contrib.pydantic import PydanticModel
class User(models.BaseUser):
pass
class UserCreate(models.BaseUserCreate):
pass
class UserUpdate(models.BaseUserUpdate):
pass
class UserModel(TortoiseBaseUserModel):
pass
class UserDB(User, models.BaseUserDB, PydanticModel):
class Config:
orm_mode = True
orig_model = UserModel

80
docs/src/full_mongodb.py
View File

@@ -1,80 +0,0 @@
import motor.motor_asyncio
from fastapi import FastAPI, Request
from fastapi_users import FastAPIUsers, models
from fastapi_users.authentication import JWTAuthentication
from fastapi_users.db import MongoDBUserDatabase
DATABASE_URL = "mongodb://localhost:27017"
SECRET = "SECRET"
class User(models.BaseUser):
pass
class UserCreate(models.BaseUserCreate):
pass
class UserUpdate(User, models.BaseUserUpdate):
pass
class UserDB(User, models.BaseUserDB):
pass
client = motor.motor_asyncio.AsyncIOMotorClient(
DATABASE_URL, uuidRepresentation="standard"
)
db = client["database_name"]
collection = db["users"]
user_db = MongoDBUserDatabase(UserDB, collection)
def on_after_register(user: UserDB, request: Request):
print(f"User {user.id} has registered.")
def on_after_forgot_password(user: UserDB, token: str, request: Request):
print(f"User {user.id} has forgot their password. Reset token: {token}")
def after_verification_request(user: UserDB, token: str, request: Request):
print(f"Verification requested for user {user.id}. Verification token: {token}")
jwt_authentication = JWTAuthentication(
secret=SECRET, lifetime_seconds=3600, tokenUrl="auth/jwt/login"
)
app = FastAPI()
fastapi_users = FastAPIUsers(
user_db,
[jwt_authentication],
User,
UserCreate,
UserUpdate,
UserDB,
)
app.include_router(
fastapi_users.get_auth_router(jwt_authentication), prefix="/auth/jwt", tags=["auth"]
)
app.include_router(
fastapi_users.get_register_router(on_after_register), prefix="/auth", tags=["auth"]
)
app.include_router(
fastapi_users.get_reset_password_router(
SECRET, after_forgot_password=on_after_forgot_password
),
prefix="/auth",
tags=["auth"],
)
app.include_router(
fastapi_users.get_verify_router(
SECRET, after_verification_request=after_verification_request
),
prefix="/auth",
tags=["auth"],
)
app.include_router(fastapi_users.get_users_router(), prefix="/users", tags=["users"])

105
docs/src/full_ormar.py
View File

@@ -1,105 +0,0 @@
import databases
import sqlalchemy
from fastapi import FastAPI, Request
from fastapi_users import FastAPIUsers, models
from fastapi_users.authentication import JWTAuthentication
from fastapi_users.db import OrmarBaseUserModel, OrmarUserDatabase
DATABASE_URL = "sqlite:///test.db"
SECRET = "SECRET"
metadata = sqlalchemy.MetaData()
database = databases.Database(DATABASE_URL)
class User(models.BaseUser):
pass
class UserCreate(models.BaseUserCreate):
pass
class UserUpdate(User, models.BaseUserUpdate):
pass
class UserDB(User, models.BaseUserDB):
pass
class UserModel(OrmarBaseUserModel):
class Meta:
tablename = "users"
metadata = metadata
database = database
engine = sqlalchemy.create_engine(DATABASE_URL)
metadata.create_all(engine)
user_db = OrmarUserDatabase(UserDB, UserModel)
def on_after_register(user: UserDB, request: Request):
print(f"User {user.id} has registered.")
def on_after_forgot_password(user: UserDB, token: str, request: Request):
print(f"User {user.id} has forgot their password. Reset token: {token}")
def after_verification_request(user: UserDB, token: str, request: Request):
print(f"Verification requested for user {user.id}. Verification token: {token}")
jwt_authentication = JWTAuthentication(
secret=SECRET, lifetime_seconds=3600, tokenUrl="auth/jwt/login"
)
app = FastAPI()
fastapi_users = FastAPIUsers(
user_db,
[jwt_authentication],
User,
UserCreate,
UserUpdate,
UserDB,
)
app.include_router(
fastapi_users.get_auth_router(jwt_authentication), prefix="/auth/jwt", tags=["auth"]
)
app.include_router(
fastapi_users.get_register_router(on_after_register), prefix="/auth", tags=["auth"]
)
app.include_router(
fastapi_users.get_reset_password_router(
SECRET, after_forgot_password=on_after_forgot_password
),
prefix="/auth",
tags=["auth"],
)
app.include_router(
fastapi_users.get_verify_router(
SECRET, after_verification_request=after_verification_request
),
prefix="/auth",
tags=["auth"],
)
app.include_router(fastapi_users.get_users_router(), prefix="/users", tags=["users"])
app.state.database = database
@app.on_event("startup")
async def startup() -> None:
database_ = app.state.database
if not database_.is_connected:
await database_.connect()
@app.on_event("shutdown")
async def shutdown() -> None:
database_ = app.state.database
if database_.is_connected:
await database_.disconnect()

101
docs/src/full_sqlalchemy.py
View File

@@ -1,101 +0,0 @@
import databases
import sqlalchemy
from fastapi import FastAPI, Request
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(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)
def on_after_register(user: UserDB, request: Request):
print(f"User {user.id} has registered.")
def on_after_forgot_password(user: UserDB, token: str, request: Request):
print(f"User {user.id} has forgot their password. Reset token: {token}")
def after_verification_request(user: UserDB, token: str, request: Request):
print(f"Verification requested for user {user.id}. Verification token: {token}")
jwt_authentication = JWTAuthentication(
secret=SECRET, lifetime_seconds=3600, tokenUrl="auth/jwt/login"
)
app = FastAPI()
fastapi_users = FastAPIUsers(
user_db,
[jwt_authentication],
User,
UserCreate,
UserUpdate,
UserDB,
)
app.include_router(
fastapi_users.get_auth_router(jwt_authentication), prefix="/auth/jwt", tags=["auth"]
)
app.include_router(
fastapi_users.get_register_router(on_after_register), prefix="/auth", tags=["auth"]
)
app.include_router(
fastapi_users.get_reset_password_router(
SECRET, after_forgot_password=on_after_forgot_password
),
prefix="/auth",
tags=["auth"],
)
app.include_router(
fastapi_users.get_verify_router(
SECRET, after_verification_request=after_verification_request
),
prefix="/auth",
tags=["auth"],
)
app.include_router(fastapi_users.get_users_router(), prefix="/users", tags=["users"])
@app.on_event("startup")
async def startup():
await database.connect()
@app.on_event("shutdown")
async def shutdown():
await database.disconnect()

88
docs/src/full_tortoise.py
View File

@@ -1,88 +0,0 @@
from fastapi import FastAPI, Request
from fastapi_users import FastAPIUsers, models
from fastapi_users.authentication import JWTAuthentication
from fastapi_users.db import TortoiseBaseUserModel, TortoiseUserDatabase
from tortoise.contrib.fastapi import register_tortoise
from tortoise.contrib.pydantic import PydanticModel
DATABASE_URL = "sqlite://./test.db"
SECRET = "SECRET"
class UserModel(TortoiseBaseUserModel):
pass
class User(models.BaseUser):
pass
class UserCreate(models.BaseUserCreate):
pass
class UserUpdate(User, models.BaseUserUpdate):
pass
class UserDB(User, models.BaseUserDB, PydanticModel):
class Config:
orm_mode = True
orig_model = UserModel
user_db = TortoiseUserDatabase(UserDB, UserModel)
app = FastAPI()
register_tortoise(
app,
db_url=DATABASE_URL,
modules={"models": ["full_tortoise"]},
generate_schemas=True,
)
def on_after_register(user: UserDB, request: Request):
print(f"User {user.id} has registered.")
def on_after_forgot_password(user: UserDB, token: str, request: Request):
print(f"User {user.id} has forgot their password. Reset token: {token}")
def after_verification_request(user: UserDB, token: str, request: Request):
print(f"Verification requested for user {user.id}. Verification token: {token}")
jwt_authentication = JWTAuthentication(
secret=SECRET, lifetime_seconds=3600, tokenUrl="auth/jwt/login"
)
fastapi_users = FastAPIUsers(
user_db,
[jwt_authentication],
User,
UserCreate,
UserUpdate,
UserDB,
)
app.include_router(
fastapi_users.get_auth_router(jwt_authentication), prefix="/auth/jwt", tags=["auth"]
)
app.include_router(
fastapi_users.get_register_router(on_after_register), prefix="/auth", tags=["auth"]
)
app.include_router(
fastapi_users.get_reset_password_router(
SECRET, after_forgot_password=on_after_forgot_password
),
prefix="/auth",
tags=["auth"],
)
app.include_router(
fastapi_users.get_verify_router(
SECRET, after_verification_request=after_verification_request
),
prefix="/auth",
tags=["auth"],
)
app.include_router(fastapi_users.get_users_router(), prefix="/users", tags=["users"])

32
docs/src/user_manager.py Normal file
View File

@@ -0,0 +1,32 @@
from typing import Optional
from fastapi import Depends, Request
from fastapi_users import BaseUserManager
from .db import get_user_db
from .models import UserCreate, UserDB
SECRET = "SECRET"
class UserManager(BaseUserManager[UserCreate, UserDB]):
user_db_model = UserDB
reset_password_token_secret = SECRET
verification_token_secret = SECRET
async def on_after_register(self, user: UserDB, request: Optional[Request] = None):
print(f"User {user.id} has registered.")
async def on_after_forgot_password(
self, user: UserDB, token: str, request: Optional[Request] = None
):
print(f"User {user.id} has forgot their password. Reset token: {token}")
async def on_after_request_verify(
self, user: UserDB, token: str, request: Optional[Request] = None
):
print(f"Verification requested for user {user.id}. Verification token: {token}")
def get_user_manager(user_db=Depends(get_user_db)):
yield UserManager(user_db)