from fastapi import FastAPI from fastapi_amis_admin.admin.settings import Settings from fastapi_user_auth.site import AuthAdminSite from starlette.requests import Request from sqlmodel import SQLModel # 创建FastAPI应用 app = FastAPI() # 创建AdminSite实例 site = AuthAdminSite(settings=Settings(database_url='sqlite:///amisadmin.db?check_same_thread=False')) auth = site.auth # 挂载后台管理系统 site.mount_app(app) # 创建初始化数据库表 @app.on_event("startup") async def startup(): await site.db.async_run_sync(SQLModel.metadata.create_all, is_session=False) # 创建默认管理员,用户名: admin,密码: admin, 请及时修改密码!!! await auth.create_role_user("admin") # 创建默认超级管理员,用户名: root,密码: root, 请及时修改密码!!! await auth.create_role_user("root") # 运行site的startup方法,加载casbin策略等 await site.router.startup() # 添加一条默认的casbin规则 if not auth.enforcer.enforce("u:admin", site.unique_id, "page", "page"): await auth.enforcer.add_policy("u:admin", site.unique_id, "page", "page", "allow") # 要求: 用户必须登录 @app.get("/auth/get_user") @auth.requires() def get_user(request: Request): return request.user if __name__ == '__main__': import uvicorn uvicorn.run(app)

# 要求: 用户必须登录 @app.get("/auth/user") @auth.requires() def user(request: Request): return request.user # 当前请求用户对象. # 验证路由: 用户拥有admin角色 @app.get("/auth/admin_roles") @auth.requires('admin') def admin_roles(request: Request): return request.user # 要求: 用户拥有vip角色 # 支持同步/异步路由 @app.get("/auth/vip_roles") @auth.requires(['vip']) async def vip_roles(request: Request): return request.user # 要求: 用户拥有admin角色 或 vip角色 @app.get("/auth/admin_or_vip_roles") @auth.requires(roles=['admin', 'vip']) def admin_or_vip_roles(request: Request): return request.user

from fastapi import Depends from fastapi_user_auth.auth.models import User # 路由参数依赖项, 推荐使用此方式 @app.get("/auth/admin_roles_depend_1") def admin_roles(user: User = Depends(auth.get_current_user)): return user # or request.user # 路径操作装饰器依赖项 @app.get("/auth/admin_roles_depend_2", dependencies=[Depends(auth.requires('admin')())]) def admin_roles(request: Request): return request.user # 全局依赖项 # 在app应用下全部请求都要求拥有admin角色 app = FastAPI(dependencies=[Depends(auth.requires('admin')())]) @app.get("/auth/admin_roles_depend_3") def admin_roles(request: Request): return request.user

app = FastAPI() # 在app应用下每条请求处理之前都附加`request.auth`和`request.user`对象 auth.backend.attach_middleware(app)

from fastapi_user_auth.auth.models import User async def get_request_user(request: Request) -> Optional[User]: # user= await auth.get_current_user(request) if await auth.requires('admin', response=False)(request): return request.user else: return None

from fastapi_user_auth.auth.backends.jwt import JwtTokenStore from sqlalchemy_database import Database from fastapi_user_auth.auth import Auth from fastapi_amis_admin.admin.site import AuthAdminSite # 创建同步数据库引擎 db = Database.create(url="sqlite:///amisadmin.db?check_same_thread=False") # 使用`JwtTokenStore`创建auth对象 auth = Auth( db=db, token_store=JwtTokenStore(secret_key='09d25e094faa6ca2556c818166b7a9563b93f7099f6f0f4caa6cf63b88e8d3e7') ) # 将auth对象传入AdminSite site = AuthAdminSite( settings=Settings(), db=db, auth=auth )

# 使用`DbTokenStore`创建auth对象 from fastapi_user_auth.auth.backends.db import DbTokenStore auth = Auth( db=db, token_store=DbTokenStore(db=db) )

# 使用`RedisTokenStore`创建auth对象 from fastapi_user_auth.auth.backends.redis import RedisTokenStore from redis.asyncio import Redis auth = Auth( db=db, token_store=RedisTokenStore(redis=Redis.from_url('redis://localhost?db=0')) )

from datetime import date from fastapi_amis_admin.models.fields import Field from fastapi_user_auth.auth.models import BaseUser # 自定义`User`模型,继承`BaseUser` class MyUser(BaseUser, table = True): point: float = Field(default = 0, title = '积分', description = '用户积分') phone: str = Field(None, title = '手机号', max_length = 15) parent_id: int = Field(None, title = "上级", foreign_key = "auth_user.id") birthday: date = Field(None, title = "出生日期") location: str = Field(None, title = "位置") # 使用自定义的`User`模型,创建auth对象 auth = Auth(db = AsyncDatabase(engine), user_model = MyUser)

from fastapi_amis_admin.models.fields import Field from fastapi_user_auth.auth.models import Role # 自定义`Role`模型,继承`Role`; class MyRole(Role, table=True): icon: str = Field(None, title='图标') is_active: bool = Field(default=True, title="是否激活")

# 自定义模型管理类,继承重写对应的默认管理类 class MyRoleAdmin(admin.ModelAdmin): page_schema = PageSchema(label='用户组管理', icon='fa fa-group') model = MyRole readonly_fields = ['key'] # 自定义用户认证应用,继承重写默认的用户认证应用 class MyUserAuthApp(UserAuthApp): RoleAdmin = MyRoleAdmin # 自定义用户管理站点,继承重写默认的用户管理站点 class MyAuthAdminSite(AuthAdminSite): UserAuthApp = MyUserAuthApp # 使用自定义的`AuthAdminSite`类,创建site对象 site = MyAuthAdminSite(settings, auth=auth)

from fastapi_user_auth.mixins.admin import AuthFieldModelAdmin from fastapi_amis_admin.amis import PageSchema from fastapi_amis_admin.admin import FieldPermEnum class AuthFieldArticleAdmin(AuthFieldModelAdmin): page_schema = PageSchema(label="文章管理") model = Article # 指定不需要权限控制的字段.  perm_fields_exclude = { FieldPermEnum.CREATE: ["title", "description", "content"], }

from fastapi_user_auth.mixins.admin import AuthSelectModelAdmin from fastapi_amis_admin.amis import PageSchema from fastapi_amis_admin.admin import RecentTimeSelectPerm, UserSelectPerm, SimpleSelectPerm class AuthSelectArticleAdmin(AuthSelectModelAdmin): page_schema = PageSchema(label="数据集控制文章管理") model = Article select_permissions = [ # 最近7天创建的数据. reverse=True表示反向选择,即默认选择最近7天之内的数据 RecentTimeSelectPerm(name="recent7_create", label="最近7天创建", td=60 * 60 * 24 * 7, reverse=True), # 最近30天创建的数据 RecentTimeSelectPerm(name="recent30_create", label="最近30天创建", td=60 * 60 * 24 * 30), # 最近3天更新的数据 RecentTimeSelectPerm(name="recent3_update", label="最近3天更新", td=60 * 60 * 24 * 3, time_column="update_time"), # 只能选择自己创建的数据, reverse=True表示反向选择,即默认选择自己创建的数据 UserSelectPerm(name="self_create", label="自己创建", user_column="user_id", reverse=True), # # 只能选择自己更新的数据 # UserSelectPerm(name="self_update", label="自己更新", user_column="update_by"), # 只能选择已发布的数据 SimpleSelectPerm(name="published", label="已发布", column="is_published", values=[True]), # 只能选择状态为[1,2,3]的数据 SimpleSelectPerm(name="status_1_2_3", label="状态为1_2_3", column="status", values=[1, 2, 3]), ]