datax-admin/app/routers/files.py

import io
import json
import time
import uuid
import app.crud as crud
from fastapi import APIRouter, File, UploadFile, Form, Depends
from sqlalchemy.orm import Session
from fastapi.responses import StreamingResponse
from utils.sx_time import sxtimeit
from utils.sx_web import web_try
from app.common.minio import minio_client

from app import get_db



router = APIRouter(
    prefix="/jpt/files",
    tags=["files-文件管理"],
)

@router.get("/")
@sxtimeit
def get_file(uri: str):
    file = minio_client.get_file(uri)
    code = 200
    if len(file) == 0:
        code = 404
    response = StreamingResponse(io.BytesIO(file), status_code=code, media_type="application/octet-stream")
    # 在请求头进行配置
    response.headers["Content-Disposition"] = "attachment; filename="+uri
    return response

@router.delete("/")
@web_try()
@sxtimeit
def delete_file(uri: str,db: Session = Depends(get_db)):
    res = crud.get_jm_homework_by_dag_url(db,uri)
    if len(res) == 0:
        minio_client.del_file(uri)
    else:
        raise Exception("该算子正在被作业使用，不可删除")

@router.post("/upload_file")
@web_try()
@sxtimeit
def upload_file(file: UploadFile = File(...), project_id: str=Form(...), file_type: str=Form(...)):
    print("UploadFile-->",file.filename)
    file_name = str(int(time.time()))+'_'+file.filename
    url = minio_client.put_byte_file(f"{project_id}/{file_type}/"+file_name, file.file.read())
    return url


@router.get("/directory")
@web_try()
@sxtimeit
def get_directory(project_id: str, user_id: str, file_type: str):
    files = minio_client.ls_file(f'{project_id}/{file_type}/')
    res = []
    for file in files:
        timestamp = file.object_name.split('_',1)[0].split('/')[-1]
        time_str = time.strftime('%Y-%m-%d %H:%M:%S',time.localtime(int(timestamp)))
        dag_name = file.object_name.split('_',1)[1]
        res.append({'name':time_str+'-'+dag_name, 'uri':file.object_name,'timestamp':timestamp})
    res.sort(key=lambda x: x['timestamp'],reverse=True)
    return res

@router.get("/dag_content")
@sxtimeit
@web_try()
def get_dag_content(uri: str):
    file = minio_client.get_file(uri)
    if len(file) == 0:
        raise Exception('No file found')
    res = json.loads(file)
    return res