105 lines
3.1 KiB
Python
105 lines
3.1 KiB
Python
from sqlalchemy.orm import Session
|
|
from sqlalchemy import asc, func
|
|
from typing import List
|
|
from fastapi.encoders import jsonable_encoder
|
|
|
|
from app.model.bussiness_model import ProjectImage as piModel, ProjectImgLabel
|
|
from app.model.schemas.project_image_schemas import ProjectImage, ProjectImagePager, ProjectImageOut
|
|
from app.db.page_util import get_pager
|
|
|
|
|
|
def get_image_pager(image: ProjectImagePager, session: Session):
|
|
query = session.query(piModel).filter(piModel.project_id == image.project_id).order_by(asc(piModel.id))
|
|
pager = get_pager(query, image.pagerNum, image.pagerSize)
|
|
pager.data = [ProjectImage.from_orm(image) for image in pager.data]
|
|
return pager
|
|
|
|
|
|
def get_image_pager2(image: ProjectImage, session: Session):
|
|
# 1 子查询
|
|
subquery = (
|
|
session.query(
|
|
ProjectImgLabel.image_id,
|
|
func.ifnull(func.count(ProjectImgLabel.id), 0).label('label_count')
|
|
)
|
|
.group_by(ProjectImgLabel.image_id)
|
|
.subquery()
|
|
)
|
|
# 2 主查询
|
|
query = (
|
|
session.query(
|
|
piModel,
|
|
func.ifnull(subquery.c.label_count, 0).label('label_count')
|
|
)
|
|
.outerjoin(subquery, piModel.id == subquery.c.image_id)
|
|
)
|
|
query = query.filter(piModel.project_id == image.project_id).order_by(asc(piModel.id))
|
|
pager = get_pager(query, image.pagerNum, image.pagerSize)
|
|
datas = []
|
|
for result in pager.data:
|
|
data = ProjectImageOut.from_orm(result[0])
|
|
data.label_count = result[1]
|
|
datas.append(data)
|
|
pager.data = jsonable_encoder(datas)
|
|
return pager
|
|
|
|
|
|
def check_img_name(project_id: int, file_name: str, session: Session):
|
|
"""
|
|
根据项目id和文件名称进行查重
|
|
:param project_id:
|
|
:param file_name:
|
|
:param session:
|
|
:return:
|
|
"""
|
|
image = session.query(piModel)\
|
|
.filter(piModel.project_id == project_id).filter(piModel.file_name == file_name).first()
|
|
if image is not None:
|
|
return False
|
|
return True
|
|
|
|
|
|
def get_img_url(image_id: int, session: Session):
|
|
"""
|
|
根据id获取图片
|
|
:param image_id:
|
|
:param session:
|
|
:return:
|
|
"""
|
|
result = session.query(piModel).filter_by(id=image_id).first()
|
|
sour_url = result.image_url
|
|
thumb_url = result.thumb_image_url
|
|
return sour_url, thumb_url
|
|
|
|
|
|
def get_image_list(project_id: int, session: Session):
|
|
query = session.query(piModel).filter(piModel.project_id == project_id).order_by(asc(piModel.id))
|
|
image_list = [ProjectImage.from_orm(image) for image in query.all()]
|
|
return image_list
|
|
|
|
|
|
def get_images(project_id: int, session: Session):
|
|
query = session.query(piModel).filter(piModel.project_id == project_id).order_by(asc(piModel.id))
|
|
return query.all()
|
|
|
|
|
|
def add_image(image: ProjectImage, session: Session):
|
|
session.add(image)
|
|
session.commit()
|
|
return image
|
|
|
|
|
|
def add_image_batch(images: List[ProjectImage], session: Session):
|
|
session.add_all(images)
|
|
session.commit()
|
|
|
|
|
|
def del_image(image_id: str, session: Session):
|
|
row_del = session.query(piModel).filter_by(id=image_id).delete()
|
|
session.commit()
|
|
return row_del
|
|
|
|
|
|
|
|
|