introduce graphql
This commit is contained in:
@@ -0,0 +1,26 @@
|
||||
from fastapi import Request
|
||||
from strawberry.fastapi import BaseContext
|
||||
from app.db import SessionLocal
|
||||
from app.auth.deps import get_current_user
|
||||
from app.models import User
|
||||
from typing import AsyncGenerator
|
||||
|
||||
class CustomContext(BaseContext):
|
||||
def __init__(self, request: Request):
|
||||
super().__init__(request)
|
||||
self.db = SessionLocal()
|
||||
self.request = request
|
||||
|
||||
async def get_current_user(self) -> User | None:
|
||||
# 复用现有的 auth 逻辑
|
||||
try:
|
||||
return await get_current_user(self.request)
|
||||
except Exception:
|
||||
return None
|
||||
|
||||
async def get_context(request: Request) -> AsyncGenerator[CustomContext, None]:
|
||||
context = CustomContext(request)
|
||||
try:
|
||||
yield context
|
||||
finally:
|
||||
context.db.close() # 确保请求结束时关闭数据库连接
|
||||
@@ -0,0 +1,18 @@
|
||||
import strawberry
|
||||
from app.auth.jwt import create_jwt_token, create_refresh_token
|
||||
|
||||
@strawberry.type
|
||||
class AuthResponse:
|
||||
access_token: str
|
||||
refresh_token: str
|
||||
token_type: str
|
||||
|
||||
@strawberry.type
|
||||
class Mutation:
|
||||
@strawberry.mutation
|
||||
def logout(self) -> str:
|
||||
return "ok"
|
||||
|
||||
# 提示:像 Google OAuth 这种涉及第三方重定向的流程,
|
||||
# 依然保留现有的 RESTful (/login/google 和 /auth/callback) 会更简单。
|
||||
# 登录成功后重定向回前端,前端拿到 token,后续所有数据交互全部走 GraphQL。
|
||||
@@ -0,0 +1,18 @@
|
||||
import strawberry
|
||||
from typing import Optional
|
||||
from app.graphql.types import UserType
|
||||
from app.models import User
|
||||
|
||||
@strawberry.type
|
||||
class Query:
|
||||
@strawberry.field
|
||||
async def me(self, info) -> Optional[UserType]:
|
||||
# 从 context 获取当前登录用户
|
||||
user = await info.context.get_current_user()
|
||||
# Strawberry 会自动将 SQLAlchemy 实例映射为 UserType
|
||||
return user
|
||||
|
||||
@strawberry.field
|
||||
def get_user_by_handle(self, info, handle: str) -> Optional[UserType]:
|
||||
db = info.context.db
|
||||
return db.query(User).filter(User.handle == handle).first()
|
||||
@@ -0,0 +1,44 @@
|
||||
import strawberry
|
||||
from typing import List, Optional
|
||||
from datetime import datetime
|
||||
from app.utils.format import get_image_url
|
||||
|
||||
@strawberry.type
|
||||
class UserType:
|
||||
id: strawberry.ID
|
||||
email: str
|
||||
name: Optional[str]
|
||||
handle: Optional[str]
|
||||
avatar: Optional[str]
|
||||
following_count: int
|
||||
follower_count: int
|
||||
created_at: datetime
|
||||
|
||||
# 动态解析字段:例如自动处理头像的绝对路径
|
||||
@strawberry.field
|
||||
def avatar_url(self) -> Optional[str]:
|
||||
return get_image_url(self.avatar) if self.avatar else None
|
||||
|
||||
@strawberry.type
|
||||
class WorkspaceType:
|
||||
id: strawberry.ID
|
||||
name: str
|
||||
slug: str
|
||||
logo: Optional[str]
|
||||
description: Optional[str]
|
||||
storage_used: int
|
||||
|
||||
# 支持前端按需加载关联的成员
|
||||
@strawberry.field
|
||||
def members(self, info) -> List["WorkspaceMemberType"]:
|
||||
from app.models import WorkspaceMember
|
||||
db = info.context.db
|
||||
return db.query(WorkspaceMember).filter(WorkspaceMember.workspace_id == self.id).all()
|
||||
|
||||
@strawberry.type
|
||||
class WorkspaceMemberType:
|
||||
id: strawberry.ID
|
||||
workspace_id: str
|
||||
user_id: Optional[str]
|
||||
role: str
|
||||
status: str
|
||||
Reference in New Issue
Block a user