mirror of
https://github.com/fastapi-practices/fastapi-best-architecture.git
synced 2026-09-21 13:12:24 +00:00
Update the OAuth2 module to plugin (#620)
* Update the OAuth2 module to plugin * update create user social param
This commit is contained in:
Wu Clan
@@ -0,0 +1,2 @@
|
||||
#!/usr/bin/env python3
|
||||
# -*- coding: utf-8 -*-
|
||||
@@ -0,0 +1,2 @@
|
||||
#!/usr/bin/env python3
|
||||
# -*- coding: utf-8 -*-
|
||||
@@ -0,0 +1,12 @@
|
||||
#!/usr/bin/env python3
|
||||
# -*- coding: utf-8 -*-
|
||||
from fastapi import APIRouter
|
||||
|
||||
from backend.core.conf import settings
|
||||
from backend.plugin.oauth2.api.v1.github import router as github_router
|
||||
from backend.plugin.oauth2.api.v1.linux_do import router as linux_do_router
|
||||
|
||||
v1 = APIRouter(prefix=f'{settings.FASTAPI_API_V1_PATH}/oauth2')
|
||||
|
||||
v1.include_router(github_router, prefix='/github', tags=['Github OAuth2'])
|
||||
v1.include_router(linux_do_router, prefix='/linux-do', tags=['LinuxDo OAuth2'])
|
||||
@@ -0,0 +1,2 @@
|
||||
#!/usr/bin/env python3
|
||||
# -*- coding: utf-8 -*-
|
||||
@@ -0,0 +1,47 @@
|
||||
#!/usr/bin/env python3
|
||||
# -*- coding: utf-8 -*-
|
||||
from fastapi import APIRouter, BackgroundTasks, Depends, Request, Response
|
||||
from fastapi_limiter.depends import RateLimiter
|
||||
from fastapi_oauth20 import FastAPIOAuth20, GitHubOAuth20
|
||||
from starlette.responses import RedirectResponse
|
||||
|
||||
from backend.common.enums import UserSocialType
|
||||
from backend.common.response.response_schema import ResponseSchemaModel, response_base
|
||||
from backend.core.conf import settings
|
||||
from backend.plugin.oauth2.service.oauth2_service import oauth2_service
|
||||
|
||||
router = APIRouter()
|
||||
|
||||
_github_client = GitHubOAuth20(settings.OAUTH2_GITHUB_CLIENT_ID, settings.OAUTH2_GITHUB_CLIENT_SECRET)
|
||||
_github_oauth2 = FastAPIOAuth20(_github_client, redirect_route_name='github_login')
|
||||
|
||||
|
||||
@router.get('', summary='获取 Github 授权链接')
|
||||
async def github_oauth2(request: Request) -> ResponseSchemaModel[str]:
|
||||
auth_url = await _github_client.get_authorization_url(redirect_uri=f'{request.url}/callback')
|
||||
return response_base.success(data=auth_url)
|
||||
|
||||
|
||||
@router.get(
|
||||
'/callback',
|
||||
summary='Github 授权自动重定向',
|
||||
description='Github 授权后，自动重定向到当前地址并获取用户信息，通过用户信息自动创建系统用户',
|
||||
dependencies=[Depends(RateLimiter(times=5, minutes=1))],
|
||||
)
|
||||
async def github_login(
|
||||
request: Request,
|
||||
response: Response,
|
||||
background_tasks: BackgroundTasks,
|
||||
oauth2: FastAPIOAuth20 = Depends(_github_oauth2),
|
||||
):
|
||||
token, _state = oauth2
|
||||
access_token = token['access_token']
|
||||
user = await _github_client.get_userinfo(access_token)
|
||||
data = await oauth2_service.create_with_login(
|
||||
request=request,
|
||||
response=response,
|
||||
background_tasks=background_tasks,
|
||||
user=user,
|
||||
social=UserSocialType.github,
|
||||
)
|
||||
return RedirectResponse(url=f'{settings.OAUTH2_FRONTEND_REDIRECT_URI}?access_token={data.access_token}')
|
||||
@@ -0,0 +1,50 @@
|
||||
#!/usr/bin/env python3
|
||||
# -*- coding: utf-8 -*-
|
||||
from fastapi import APIRouter, BackgroundTasks, Depends, Request, Response
|
||||
from fastapi_limiter.depends import RateLimiter
|
||||
from fastapi_oauth20 import FastAPIOAuth20, LinuxDoOAuth20
|
||||
from starlette.responses import RedirectResponse
|
||||
|
||||
from backend.common.enums import UserSocialType
|
||||
from backend.common.response.response_schema import ResponseSchemaModel, response_base
|
||||
from backend.core.conf import settings
|
||||
from backend.plugin.oauth2.service.oauth2_service import oauth2_service
|
||||
|
||||
router = APIRouter()
|
||||
|
||||
_linux_do_client = LinuxDoOAuth20(
|
||||
settings.OAUTH2_LINUX_DO_CLIENT_ID,
|
||||
settings.OAUTH2_LINUX_DO_CLIENT_SECRET,
|
||||
)
|
||||
_linux_do_oauth2 = FastAPIOAuth20(_linux_do_client, redirect_route_name='linux_do_login')
|
||||
|
||||
|
||||
@router.get('', summary='获取 LinuxDo 授权链接')
|
||||
async def linux_do_oauth2(request: Request) -> ResponseSchemaModel[str]:
|
||||
auth_url = await _linux_do_client.get_authorization_url(redirect_uri=f'{request.url}/callback')
|
||||
return response_base.success(data=auth_url)
|
||||
|
||||
|
||||
@router.get(
|
||||
'/callback',
|
||||
summary='LinuxDo 授权自动重定向',
|
||||
description='LinuxDo 授权后，自动重定向到当前地址并获取用户信息，通过用户信息自动创建系统用户',
|
||||
dependencies=[Depends(RateLimiter(times=5, minutes=1))],
|
||||
)
|
||||
async def linux_do_login(
|
||||
request: Request,
|
||||
response: Response,
|
||||
background_tasks: BackgroundTasks,
|
||||
oauth2: FastAPIOAuth20 = Depends(_linux_do_oauth2),
|
||||
):
|
||||
token, _state = oauth2
|
||||
access_token = token['access_token']
|
||||
user = await _linux_do_client.get_userinfo(access_token)
|
||||
data = await oauth2_service.create_with_login(
|
||||
request=request,
|
||||
response=response,
|
||||
background_tasks=background_tasks,
|
||||
user=user,
|
||||
social=UserSocialType.linux_do,
|
||||
)
|
||||
return RedirectResponse(url=f'{settings.OAUTH2_FRONTEND_REDIRECT_URI}?access_token={data.access_token}')
|
||||
@@ -0,0 +1,2 @@
|
||||
#!/usr/bin/env python3
|
||||
# -*- coding: utf-8 -*-
|
||||
@@ -0,0 +1,45 @@
|
||||
#!/usr/bin/env python3
|
||||
# -*- coding: utf-8 -*-
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
from sqlalchemy_crud_plus import CRUDPlus
|
||||
|
||||
from backend.plugin.oauth2.model import UserSocial
|
||||
from backend.plugin.oauth2.schema.user_social import CreateUserSocialParam
|
||||
|
||||
|
||||
class CRUDUserSocial(CRUDPlus[UserSocial]):
|
||||
"""用户社交账号数据库操作类"""
|
||||
|
||||
async def get(self, db: AsyncSession, pk: int, source: str) -> UserSocial | None:
|
||||
"""
|
||||
获取用户社交账号绑定详情
|
||||
|
||||
:param db: 数据库会话
|
||||
:param pk: 用户 ID
|
||||
:param source: 社交账号类型
|
||||
:return:
|
||||
"""
|
||||
return await self.select_model_by_column(db, user_id=pk, source=source)
|
||||
|
||||
async def create(self, db: AsyncSession, obj: CreateUserSocialParam) -> None:
|
||||
"""
|
||||
创建用户社交账号绑定
|
||||
|
||||
:param db: 数据库会话
|
||||
:param obj: 创建用户社交账号绑定参数
|
||||
:return:
|
||||
"""
|
||||
await self.create_model(db, obj)
|
||||
|
||||
async def delete(self, db: AsyncSession, social_id: int) -> int:
|
||||
"""
|
||||
删除用户社交账号绑定
|
||||
|
||||
:param db: 数据库会话
|
||||
:param social_id: 社交账号绑定 ID
|
||||
:return:
|
||||
"""
|
||||
return await self.delete_model(db, social_id)
|
||||
|
||||
|
||||
user_social_dao: CRUDUserSocial = CRUDUserSocial(UserSocial)
|
||||
@@ -0,0 +1,3 @@
|
||||
#!/usr/bin/env python3
|
||||
# -*- coding: utf-8 -*-
|
||||
from backend.plugin.oauth2.model.user_social import UserSocial
|
||||
@@ -0,0 +1,33 @@
|
||||
#!/usr/bin/env python3
|
||||
# -*- coding: utf-8 -*-
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import TYPE_CHECKING
|
||||
|
||||
from sqlalchemy import ForeignKey, String
|
||||
from sqlalchemy.orm import Mapped, mapped_column, relationship
|
||||
|
||||
from backend.common.model import Base, id_key
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from backend.app.admin.model import User
|
||||
|
||||
|
||||
class UserSocial(Base):
|
||||
"""用户社交表（OAuth2）"""
|
||||
|
||||
__tablename__ = 'sys_user_social'
|
||||
|
||||
id: Mapped[id_key] = mapped_column(init=False)
|
||||
source: Mapped[str] = mapped_column(String(20), comment='第三方用户来源')
|
||||
open_id: Mapped[str | None] = mapped_column(String(20), default=None, comment='第三方用户 open id')
|
||||
sid: Mapped[str | None] = mapped_column(String(20), default=None, comment='第三方用户 ID')
|
||||
union_id: Mapped[str | None] = mapped_column(String(20), default=None, comment='第三方用户 union id')
|
||||
scope: Mapped[str | None] = mapped_column(String(120), default=None, comment='第三方用户授予的权限')
|
||||
code: Mapped[str | None] = mapped_column(String(50), default=None, comment='用户的授权 code')
|
||||
|
||||
# 用户社交信息一对多
|
||||
user_id: Mapped[int | None] = mapped_column(
|
||||
ForeignKey('sys_user.id', ondelete='SET NULL'), default=None, comment='用户关联ID'
|
||||
)
|
||||
user: Mapped[User | None] = relationship(init=False, backref='socials')
|
||||
@@ -0,0 +1,8 @@
|
||||
[plugin]
|
||||
summary = 'OAuth 2.0'
|
||||
version = '0.0.1'
|
||||
description = '通过 OAuth 2.0 的方式登录系统'
|
||||
author = 'wu-clan'
|
||||
|
||||
[app]
|
||||
router = ['v1']
|
||||
@@ -0,0 +1 @@
|
||||
fastapi-oauth20>=0.0.1
|
||||
@@ -0,0 +1,2 @@
|
||||
#!/usr/bin/env python3
|
||||
# -*- coding: utf-8 -*-
|
||||
@@ -0,0 +1,27 @@
|
||||
#!/usr/bin/env python3
|
||||
# -*- coding: utf-8 -*-
|
||||
from pydantic import Field
|
||||
|
||||
from backend.common.enums import UserSocialType
|
||||
from backend.common.schema import SchemaBase
|
||||
|
||||
|
||||
class UserSocialSchemaBase(SchemaBase):
|
||||
"""用户社交基础模型"""
|
||||
|
||||
source: UserSocialType = Field(description='社交平台')
|
||||
open_id: str | None = Field(None, description='开放平台 ID')
|
||||
sid: str | None = Field(None, description='第三方用户 ID')
|
||||
union_id: str | None = Field(None, description='开放平台唯一 ID')
|
||||
scope: str | None = Field(None, description='授权范围')
|
||||
code: str | None = Field(None, description='授权码')
|
||||
|
||||
|
||||
class CreateUserSocialParam(UserSocialSchemaBase):
|
||||
"""创建用户社交参数"""
|
||||
|
||||
user_id: int = Field(description='用户 ID')
|
||||
|
||||
|
||||
class UpdateUserSocialParam(SchemaBase):
|
||||
"""更新用户社交参数"""
|
||||
@@ -0,0 +1,2 @@
|
||||
#!/usr/bin/env python3
|
||||
# -*- coding: utf-8 -*-
|
||||
@@ -0,0 +1,124 @@
|
||||
#!/usr/bin/env python3
|
||||
# -*- coding: utf-8 -*-
|
||||
from typing import Any
|
||||
|
||||
from fast_captcha import text_captcha
|
||||
from fastapi import BackgroundTasks, Request, Response
|
||||
|
||||
from backend.app.admin.crud.crud_user import user_dao
|
||||
from backend.app.admin.schema.token import GetLoginToken
|
||||
from backend.app.admin.schema.user import RegisterUserParam
|
||||
from backend.app.admin.service.login_log_service import login_log_service
|
||||
from backend.common.enums import LoginLogStatusType, UserSocialType
|
||||
from backend.common.exception.errors import AuthorizationError
|
||||
from backend.common.security import jwt
|
||||
from backend.core.conf import settings
|
||||
from backend.database.db import async_db_session
|
||||
from backend.database.redis import redis_client
|
||||
from backend.plugin.oauth2.crud.crud_user_social import user_social_dao
|
||||
from backend.plugin.oauth2.schema.user_social import CreateUserSocialParam
|
||||
from backend.utils.timezone import timezone
|
||||
|
||||
|
||||
class OAuth2Service:
|
||||
"""OAuth2 认证服务类"""
|
||||
|
||||
@staticmethod
|
||||
async def create_with_login(
|
||||
*,
|
||||
request: Request,
|
||||
response: Response,
|
||||
background_tasks: BackgroundTasks,
|
||||
user: dict[str, Any],
|
||||
social: UserSocialType,
|
||||
) -> GetLoginToken | None:
|
||||
"""
|
||||
创建 OAuth2 用户并登录
|
||||
|
||||
:param request: FastAPI 请求对象
|
||||
:param response: FastAPI 响应对象
|
||||
:param background_tasks: FastAPI 后台任务
|
||||
:param user: OAuth2 用户信息
|
||||
:param social: 社交平台类型
|
||||
:return:
|
||||
"""
|
||||
async with async_db_session.begin() as db:
|
||||
# 获取 OAuth2 平台用户信息
|
||||
social_id = user.get('id')
|
||||
social_nickname = user.get('name')
|
||||
|
||||
social_username = user.get('username')
|
||||
if social == UserSocialType.github:
|
||||
social_username = user.get('login')
|
||||
|
||||
social_email = user.get('email')
|
||||
if social == UserSocialType.linux_do:
|
||||
social_email = f'{social_username}@linux.do'
|
||||
if not social_email:
|
||||
raise AuthorizationError(msg=f'授权失败，{social.value} 账户未绑定邮箱')
|
||||
|
||||
# 创建系统用户
|
||||
sys_user = await user_dao.check_email(db, social_email)
|
||||
if not sys_user:
|
||||
sys_user = await user_dao.get_by_username(db, social_username)
|
||||
if sys_user:
|
||||
social_username = f'{social_username}#{text_captcha(5)}'
|
||||
sys_user = await user_dao.get_by_nickname(db, social_nickname)
|
||||
if sys_user:
|
||||
social_username = f'{social_nickname}#{text_captcha(5)}'
|
||||
new_sys_user = RegisterUserParam(
|
||||
username=social_username, password=None, nickname=social_username, email=social_email
|
||||
)
|
||||
await user_dao.create(db, new_sys_user, social=True)
|
||||
await db.flush()
|
||||
sys_user = await user_dao.check_email(db, social_email)
|
||||
# 绑定社交用户
|
||||
sys_user_id = sys_user.id
|
||||
user_social = await user_social_dao.get(db, sys_user_id, social.value)
|
||||
if not user_social:
|
||||
new_user_social = CreateUserSocialParam(source=social.value, sid=str(social_id), user_id=sys_user_id)
|
||||
await user_social_dao.create(db, new_user_social)
|
||||
# 创建 token
|
||||
access_token = await jwt.create_access_token(
|
||||
str(sys_user_id),
|
||||
sys_user.is_multi_login,
|
||||
# extra info
|
||||
username=sys_user.username,
|
||||
nickname=sys_user.nickname,
|
||||
last_login_time=timezone.t_str(timezone.now()),
|
||||
ip=request.state.ip,
|
||||
os=request.state.os,
|
||||
browser=request.state.browser,
|
||||
device=request.state.device,
|
||||
)
|
||||
refresh_token = await jwt.create_refresh_token(str(sys_user_id), multi_login=sys_user.is_multi_login)
|
||||
await user_dao.update_login_time(db, sys_user.username)
|
||||
await db.refresh(sys_user)
|
||||
login_log = dict(
|
||||
db=db,
|
||||
request=request,
|
||||
user_uuid=sys_user.uuid,
|
||||
username=sys_user.username,
|
||||
login_time=timezone.now(),
|
||||
status=LoginLogStatusType.success.value,
|
||||
msg='登录成功（OAuth2）',
|
||||
)
|
||||
background_tasks.add_task(login_log_service.create, **login_log)
|
||||
await redis_client.delete(f'{settings.CAPTCHA_LOGIN_REDIS_PREFIX}:{request.state.ip}')
|
||||
response.set_cookie(
|
||||
key=settings.COOKIE_REFRESH_TOKEN_KEY,
|
||||
value=refresh_token.refresh_token,
|
||||
max_age=settings.COOKIE_REFRESH_TOKEN_EXPIRE_SECONDS,
|
||||
expires=timezone.f_utc(refresh_token.refresh_token_expire_time),
|
||||
httponly=True,
|
||||
)
|
||||
data = GetLoginToken(
|
||||
access_token=access_token.access_token,
|
||||
access_token_expire_time=access_token.access_token_expire_time,
|
||||
user=sys_user, # type: ignore
|
||||
session_uuid=access_token.session_uuid,
|
||||
)
|
||||
return data
|
||||
|
||||
|
||||
oauth2_service: OAuth2Service = OAuth2Service()
|
||||
@@ -149,9 +149,9 @@ def inject_extra_router(plugin: dict[str, Any]) -> None:
|
||||
continue
|
||||
|
||||
# 解析插件路由配置
|
||||
file_config = plugin.get('api', {}).get(f'{file[:-3]}', {})
|
||||
prefix = file_config.get('prefix', '')
|
||||
tags = file_config.get('tags', [])
|
||||
file_config = plugin['api'][file[:-3]]
|
||||
prefix = file_config['prefix']
|
||||
tags = file_config['tags']
|
||||
|
||||
# 获取插件路由模块
|
||||
file_path = os.path.join(root, file)
|
||||
@@ -204,7 +204,7 @@ def inject_app_router(plugin: dict[str, Any], target_router: APIRouter) -> None:
|
||||
module_path = f'backend.plugin.{plugin_name}.api.router'
|
||||
try:
|
||||
module = import_module_cached(module_path)
|
||||
routers = plugin.get('app', {}).get('router')
|
||||
routers = plugin['app']['router']
|
||||
if not routers or not isinstance(routers, list):
|
||||
raise PluginConfigError(f'应用级插件 {plugin_name} 配置文件存在错误，请检查')
|
||||
|
||||
|
||||
Reference in New Issue
Block a user