网站服务器设置_邢台度网网站建设_怎么让百度收录自己的网站_上海最新新闻热点事件

FastAPI-Users是一个用于用户身份管理的扩展。它简化了用户注册、登录和权限管理的流程，用户只需初始化FastAPIUsers并包含路由即可。适用于需要完整用户管理系统（注册/登录/权限）的场景。

依赖下载

Python 3.9.21

pip install fastapi uvicorn

pip install pydantic

pip install aiosqlite

pip install fastapi-users

pip install fastapi-users[sqlalchemy] python-jose[cryptography]

pip install pyjwt

pip install passlib

pip install bcrypt

代码说明

目录结构

Main.py

import uuidfrom fastapi import FastAPI, HTTPException, status, Depends, Queryfrom fastapi.responses import JSONResponsefrom fastapi.middleware.cors import CORSMiddleware"""Import our toolsThis is the database connection file"""from db import enginefrom config import settingsfrom models.users import User, UserCreate, UserRead, UserUpdate, auth_backend, current_active_user, fastapi_usersapp = FastAPI(title=settings.PROJECT_NAME,version=settings.PROJECT_VERSION)"""Setup our origins......for now it's just our local environments"""origins = ["http://localhost:5173"]"""Add the CORS middleware, it will pass the proper CORS headers- https://fastapi.tiangolo.com/tutorial/middleware/- https://fastapi.tiangolo.com/tutorial/cors/"""app.add_middleware(CORSMiddleware,allow_origins=origins,allow_credentials=True,allow_methods=["*"],allow_headers=["*"],)"""FastAPI Users is going to give us all of our auth routes,The `prefix` is the url that will precede each url.These settings will generate these routes:- auth/register- auth/login- auth/logout"""app.include_router(fastapi_users.get_auth_router(auth_backend),prefix="/auth",tags=["auth"],)app.include_router(fastapi_users.get_register_router(UserRead, UserCreate),prefix="/auth",tags=["auth"],)"""We now return you to your regular routes."""@app.get("/")def home():return {"message": "Root Route"}"""This is a demo protected route, it will onlyreturn the response if a user is authenticated."""@app.get("/authenticated-route-demo")async def authenticated_route(user: User = Depends(current_active_user)):return {"message": f"Hello {user.email}!"}

Db_connect.py

from sqlalchemy import create_enginefrom sqlalchemy.engine import URLfrom sqlalchemy.orm import sessionmakerurl = URL.create(drivername="",username="",password="",host="localhost",database="",port=5432)engine = create_engine(url)Session = sessionmaker(bind=engine)session = Session()

Db.py

from collections.abc import AsyncGeneratorfrom fastapi import Dependsfrom fastapi_users.db import SQLAlchemyBaseUserTableUUID, SQLAlchemyUserDatabasefrom fastapi_users_db_sqlalchemy.access_token import (SQLAlchemyAccessTokenDatabase,SQLAlchemyBaseAccessTokenTableUUID,)from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_enginefrom sqlalchemy.orm import DeclarativeBasefrom config import settingsDATABASE_URL = f"sqlite+aiosqlite:///example.db"class Base(DeclarativeBase):passclass User(SQLAlchemyBaseUserTableUUID, Base):passclass AccessToken(SQLAlchemyBaseAccessTokenTableUUID, Base):passengine = create_async_engine(DATABASE_URL)async_session_maker = async_sessionmaker(engine, expire_on_commit=False)async def create_db_and_tables():async with engine.begin() as conn:await conn.run_sync(Base.metadata.create_all)async def get_async_session() -> AsyncGenerator[AsyncSession, None]:async with async_session_maker() as session:yield sessionasync def get_user_db(session: AsyncSession = Depends(get_async_session)):yield SQLAlchemyUserDatabase(session, User)async def get_access_token_db(session: AsyncSession = Depends(get_async_session),):yield SQLAlchemyAccessTokenDatabase(session, AccessToken)

Config.py

from fastapi import HTTPException, statusimport osimport sysimport base64import jsonfrom dotenv import load_dotenvfrom pathlib import Pathenv_path = Path('.') / '.env'load_dotenv(dotenv_path=env_path)def decode(variable):"""Decodes a Platform.sh environment variable.Args:variable (string):Base64-encoded JSON (the content of an environment variable).Returns:An dict (if representing a JSON object), or a scalar type.Raises:JSON decoding error.Taken from:https://github.com/platformsh-templates/django4/blob/master/myapp/settings.py"""try:if sys.version_info[1] > 5:return json.loads(base64.b64decode(variable))else:return json.loads(base64.b64decode(variable).decode('utf-8'))except json.decoder.JSONDecodeError:print('Error decoding JSON, code %d', json.decoder.JSONDecodeError)class Settings:PROJECT_NAME: str = "FastAPI Users Test"PROJECT_VERSION: str = "1.0.0"PLATFORMSH_DB_RELATIONSHIP = 'postgresqldatabase'# Import some Platform.sh settings from the environment.if (os.getenv('PLATFORM_APPLICATION_NAME') is not None):DEBUG = Falseif (os.getenv('PLATFORM_APP_DIR') is not None):STATIC_ROOT = os.path.join(os.getenv('PLATFORM_APP_DIR'), 'static')if (os.getenv('PLATFORM_PROJECT_ENTROPY') is not None):SECRET_KEY = os.getenv('PLATFORM_PROJECT_ENTROPY')# Database service configuration, post-build only.if (os.getenv('PLATFORM_ENVIRONMENT') is not None):# Using Platform.shplatform_relationships = decode(os.environ.get("PLATFORM_RELATIONSHIPS", "{}"))db_settings = platform_relationships[PLATFORMSH_DB_RELATIONSHIP][0]POSTGRES_HOST = db_settings["host"]POSTGRES_PORT = db_settings["port"]POSTGRES_USER = db_settings["username"]POSTGRES_PASSWORD = db_settings['password']POSTGRES_DB = db_settings['path']else:POSTGRES_USER: str = os.getenv("POSTGRES_USER")POSTGRES_PASSWORD = os.getenv("POSTGRES_PASSWORD")POSTGRES_HOST: str = os.getenv("POSTGRES_SERVER", "localhost")POSTGRES_PORT: str = os.getenv("POSTGRES_PORT", 5432)POSTGRES_DB: str = os.getenv("POSTGRES_DB", "tdd")SECRET_KEY = "09d25e094faa6ca2556c818166b7a9563b93f7099f6f0f4caa6cf63b88e8d3e7"ALGORITHM = "HS256"DATABASE_URL = f"sqlite+aiosqlite:///example.db"settings = Settings()

Users.py

import uuidfrom typing import Optionalfrom fastapi import Depends, Requestfrom fastapi_users import FastAPIUsers, schemas, BaseUserManager, UUIDIDMixinfrom fastapi_users.db import SQLAlchemyBaseUserTableUUID, SQLAlchemyUserDatabasefrom fastapi_users.authentication import (AuthenticationBackend,BearerTransport,JWTStrategy,)from fastapi_users_db_sqlalchemy.access_token import (SQLAlchemyAccessTokenDatabase,SQLAlchemyBaseAccessTokenTableUUID,)from sqlalchemy.ext.asyncio import AsyncSessionfrom sqlalchemy.orm import DeclarativeBasefrom config import settingsfrom db import get_async_sessionbearer_transport = BearerTransport(tokenUrl="/auth/login")class Base(DeclarativeBase):passclass User(SQLAlchemyBaseUserTableUUID, Base):passclass UserRead(schemas.BaseUser[uuid.UUID]):passclass UserCreate(schemas.BaseUserCreate):passclass UserUpdate(schemas.BaseUserUpdate):passclass UserManager(UUIDIDMixin, BaseUserManager[User, uuid.UUID]):reset_password_token_secret = settings.SECRET_KEYverification_token_secret = settings.SECRET_KEYasync def on_after_register(self, user: User, request: Optional[Request] = None):print(f"User {user.id} has registered.")async def on_after_forgot_password(self, user: User, 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: User, token: str, request: Optional[Request] = None):print(f"Verification requested for user {user.id}. Verification token: {token}")class AccessToken(SQLAlchemyBaseAccessTokenTableUUID, Base):pass"""I'm including the following items here and importing them into main.py- The docs put the schemas in their own file- The docs put the get_user_db into the db.py file"""async def get_user_db(session: AsyncSession = Depends(get_async_session)):yield SQLAlchemyUserDatabase(session, User)async def get_user_manager(user_db=Depends(get_user_db)):yield UserManager(user_db)async def get_access_token_db(session: AsyncSession = Depends(get_async_session),):yield SQLAlchemyAccessTokenDatabase(session, AccessToken)"""This sets up our JWT for each login.FastAPI Users takes care of managing these for us,including disabling on logout."""def get_jwt_strategy() -> JWTStrategy:return JWTStrategy(secret=settings.SECRET_KEY, lifetime_seconds=3600)auth_backend = AuthenticationBackend(name="jwt",transport=bearer_transport,get_strategy=get_jwt_strategy,)fastapi_users = FastAPIUsers[User, uuid.UUID](get_user_manager, [auth_backend])current_active_user = fastapi_users.current_user(active=True)