mirror of
https://github.com/fastapi-practices/fastapi_best_architecture.git
synced 2025-08-17 13:54:14 +08:00
3b64aeb97b
* [WIP] Add OAuth 2.0 authorization login * Add social user relationship table * Update social user relationship table back_populates * Add OAuth 2.0 related interface * Automatically redirect authorization addresses * Update OAuth2 authorization to GitHub * Add implementation code * fix the callback interface return * fix typo * fix the api return * fix imports * Fix logic for creating system users and social tables * Fix user information storage * Add OAuth2 source link * remove unnecessary db refresh * remove the front end docker-compose annotation
26 lines
1.0 KiB
Python
26 lines
1.0 KiB
Python
#!/usr/bin/env python3
|
|
# -*- coding: utf-8 -*-
|
|
from sqlalchemy import and_, select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from backend.app.common.enums import UserSocialType
|
|
from backend.app.crud.base import CRUDBase
|
|
from backend.app.models import UserSocial
|
|
from backend.app.schemas.user_social import CreateUserSocialParam, UpdateUserSocialParam
|
|
|
|
|
|
class CRUDOUserSocial(CRUDBase[UserSocial, CreateUserSocialParam, UpdateUserSocialParam]):
|
|
async def get(self, db: AsyncSession, pk: int, source: UserSocialType) -> UserSocial | None:
|
|
se = select(self.model).where(and_(self.model.id == pk, self.model.source == source))
|
|
user_social = await db.execute(se)
|
|
return user_social.scalars().first()
|
|
|
|
async def create(self, db: AsyncSession, obj_in: CreateUserSocialParam) -> None:
|
|
await self.create_(db, obj_in)
|
|
|
|
async def delete(self, db: AsyncSession, social_id: int) -> int:
|
|
return await self.delete_(db, social_id)
|
|
|
|
|
|
user_social_dao: CRUDOUserSocial = CRUDOUserSocial(UserSocial)
|