57 lines
1.6 KiB
Python
57 lines
1.6 KiB
Python
import time
|
|
|
|
from sqlalchemy import select
|
|
|
|
from chongming_worker import rpc_method
|
|
|
|
from ..models.pydantic import (
|
|
MeGetRequest,
|
|
MeGetResponse
|
|
)
|
|
from ..models.auth import Users
|
|
|
|
|
|
@rpc_method(
|
|
subject="auth.me",
|
|
method="GET",
|
|
summary="获取当前用户信息",
|
|
response_model=MeGetResponse,
|
|
)
|
|
async def me(me_get_request: MeGetRequest) -> MeGetResponse:
|
|
from ..global_var import db
|
|
|
|
if db is None:
|
|
raise RuntimeError("Database not initialized")
|
|
|
|
async with db.get_session() as session:
|
|
stmt = select(Users).filter_by(id=me_get_request.user_id)
|
|
result = await session.execute(stmt)
|
|
user: Users | None = result.scalar_one_or_none()
|
|
|
|
if user is None:
|
|
return MeGetResponse(
|
|
result="fail",
|
|
id="",
|
|
username="",
|
|
email="",
|
|
is_active=False,
|
|
is_verified=False,
|
|
is_superuser=False,
|
|
)
|
|
|
|
return MeGetResponse(
|
|
result="success",
|
|
id=user.id,
|
|
username=user.username,
|
|
email=user.email,
|
|
full_name=user.full_name,
|
|
avatar_url=user.avatar_url,
|
|
phone=user.phone,
|
|
is_active=user.is_active,
|
|
is_verified=user.is_verified,
|
|
is_superuser=user.is_superuser,
|
|
created_at=user.created_at.isoformat() if user.created_at else None,
|
|
updated_at=user.updated_at.isoformat() if user.updated_at else None,
|
|
last_login_at=user.last_login_at.isoformat() if user.last_login_at else None,
|
|
)
|