94 lines
3.2 KiB
Python
94 lines
3.2 KiB
Python
from sqlalchemy.ext.asyncio import AsyncSession
|
|
from sqlalchemy.future import select
|
|
from sqlalchemy import delete, update
|
|
from app.models.user_images import UserImage
|
|
from app.schemas.user_image import UserImageCreate, UserImageUpdate
|
|
from typing import List, Optional
|
|
|
|
async def get_user_image(db: AsyncSession, image_id: int):
|
|
"""获取单个用户形象"""
|
|
result = await db.execute(select(UserImage).filter(UserImage.id == image_id))
|
|
return result.scalars().first()
|
|
|
|
async def get_user_images_by_user(db: AsyncSession, user_id: int, skip: int = 0, limit: int = 100):
|
|
"""获取用户的所有形象图片"""
|
|
result = await db.execute(
|
|
select(UserImage)
|
|
.filter(UserImage.user_id == user_id)
|
|
.order_by(UserImage.create_time.desc())
|
|
.offset(skip)
|
|
.limit(limit)
|
|
)
|
|
return result.scalars().all()
|
|
|
|
async def create_user_image(db: AsyncSession, image: UserImageCreate):
|
|
"""创建用户形象
|
|
|
|
image: 包含user_id的UserImageCreate对象
|
|
"""
|
|
# 如果设置为默认形象，先重置用户的所有形象为非默认
|
|
if image.is_default and image.user_id:
|
|
await reset_default_images(db, image.user_id)
|
|
|
|
db_image = UserImage(
|
|
user_id=image.user_id,
|
|
image_url=image.image_url,
|
|
is_default=image.is_default
|
|
)
|
|
db.add(db_image)
|
|
await db.commit()
|
|
await db.refresh(db_image)
|
|
return db_image
|
|
|
|
async def update_user_image(db: AsyncSession, image_id: int, image: UserImageUpdate):
|
|
"""更新用户形象"""
|
|
db_image = await get_user_image(db, image_id)
|
|
if not db_image:
|
|
return None
|
|
|
|
# 处理默认形象逻辑
|
|
if image.is_default is not None and image.is_default and not db_image.is_default:
|
|
# 如果设置为默认且之前不是默认，重置其他形象
|
|
await reset_default_images(db, db_image.user_id)
|
|
db_image.is_default = True
|
|
elif image.is_default is not None:
|
|
db_image.is_default = image.is_default
|
|
|
|
# 更新图片URL
|
|
if image.image_url:
|
|
db_image.image_url = image.image_url
|
|
|
|
await db.commit()
|
|
await db.refresh(db_image)
|
|
return db_image
|
|
|
|
async def delete_user_image(db: AsyncSession, image_id: int):
|
|
"""删除用户形象"""
|
|
db_image = await get_user_image(db, image_id)
|
|
if db_image:
|
|
await db.delete(db_image)
|
|
await db.commit()
|
|
return db_image
|
|
|
|
async def delete_user_images(db: AsyncSession, user_id: int):
|
|
"""删除用户所有形象图片"""
|
|
stmt = delete(UserImage).where(UserImage.user_id == user_id)
|
|
await db.execute(stmt)
|
|
await db.commit()
|
|
return True
|
|
|
|
async def reset_default_images(db: AsyncSession, user_id: int):
|
|
"""重置用户所有形象为非默认"""
|
|
stmt = update(UserImage).where(
|
|
UserImage.user_id == user_id,
|
|
UserImage.is_default == True
|
|
).values(is_default=False)
|
|
await db.execute(stmt)
|
|
|
|
async def get_default_image(db: AsyncSession, user_id: int):
|
|
"""获取用户的默认形象"""
|
|
result = await db.execute(
|
|
select(UserImage)
|
|
.filter(UserImage.user_id == user_id, UserImage.is_default == True)
|
|
)
|
|
return result.scalars().first() |