46 lines
1.3 KiB
Python
46 lines
1.3 KiB
Python
|
from typing import List
|
||
|
|
|
||
|
|
from .Model import FormFile
|
||
|
|
from sqlalchemy.orm import Session
|
||
|
|
from . import Sechemas
|
||
|
|
|
||
|
|
|
||
|
|
def get_file_by_md(db: Session, md: str):
|
||
|
|
item = db.query(FormFile).filter_by(md=md).first()
|
||
|
|
return item
|
||
|
|
|
||
|
|
|
||
|
|
def add_file(db: Session, data: Sechemas.FormFileCreate):
|
||
|
|
new_file = FormFile(**data.dict())
|
||
|
|
db.add(new_file)
|
||
|
|
db.commit()
|
||
|
|
db.refresh(new_file)
|
||
|
|
return new_file
|
||
|
|
|
||
|
|
|
||
|
|
def file_get(db: Session, md: str):
|
||
|
|
file = db.query(FormFile).filter_by(md=md).first()
|
||
|
|
return file
|
||
|
|
|
||
|
|
|
||
|
|
def file_query(db: Session, params: Sechemas.FormFileQuery) -> (int, List[FormFile]):
|
||
|
|
query = db.query(FormFile)
|
||
|
|
if params.md:
|
||
|
|
query.filter(FormFile.md == params.md)
|
||
|
|
if params.name:
|
||
|
|
query.filter(FormFile.name.like(f'%{params.name}%'))
|
||
|
|
if params.user_id:
|
||
|
|
query.filter(FormFile.user_id == params.user_id)
|
||
|
|
if params.common_type:
|
||
|
|
query.filter(FormFile.common_type == params.common_type)
|
||
|
|
if params.mime_type:
|
||
|
|
query.filter(FormFile.mime_type == params.mime_type)
|
||
|
|
if params.file_url:
|
||
|
|
query.filter(FormFile.file_url == params.file_url)
|
||
|
|
count = query.count()
|
||
|
|
page = params.page
|
||
|
|
page_size = params.page_size
|
||
|
|
if page is not None and page_size is not None:
|
||
|
|
query = query.offset((page - 1) * page_size).limit(page_size).all()
|
||
|
|
return count, query
|