datax-admin/app/crud/data_management.py

from re import S
import time
from typing import List
from app import models, schemas
from sqlalchemy.orm import Session

def create_data_management(db: Session, item: schemas.DataManagementCreate, table_name: str, af_run_id: str):
    create_time: int = int(time.time())
    db_item = models.DataManagement(**{
        'name': item.name,
        'table_name': table_name,
        'create_time': create_time,
        'user_name': item.user_name,
        'user_id': item.user_id,
        'project_id': item.project_id,
        'af_run_id': af_run_id,
        'status': 1
    })
    db.add(db_item)
    db.commit()
    db.refresh(db_item)
    return db_item

def update_data_management_status(db: Session, d_id: int, status: int):
    db_item: models.DataManagement = db.query(models.DataManagement).filter(models.DataManagement.id == d_id).first()
    db_item.status = status
    db.commit()
    db.flush()
    db.refresh(db_item)
    return db_item

def get_data_managements(db: Session, user_id: str, project_id: str):
    res: List[models.DataManagement] = db.query(models.DataManagement).filter(models.DataManagement.project_id == project_id).all()
    return res

def get_data_management_info(db: Session, d_id: int):
    db_item: models.DataManagement = db.query(models.DataManagement).filter(models.DataManagement.id == d_id).first()
    if not db_item:
        raise Exception('该数据不存在')
    return db_item

def delete_data_management(db: Session, d_id: int):
    dm_item = db.query(models.DataManagement).filter(models.DataManagement.id == d_id).first()
    if not dm_item:
        raise Exception("Data management not found")
    db.delete(dm_item)
    db.commit()
    db.flush()
    return dm_item