datax-admin/app/crud/datax_json.py

from asyncio.log import logger
import time
from typing import List
from app import models, schemas
from sqlalchemy.orm import Session
from app.core.datax import *
from configs.logging import logger


def generate_datax_json(db: Session, param: schemas.DataXJsonParam):
    reader_ds = db.query(models.JobJdbcDatasource).filter(models.JobJdbcDatasource.id == param.reader_datasource_id).first()
    if not reader_ds:
        raise Exception('Reader datasource not found')
    writer_ds = db.query(models.JobJdbcDatasource).filter(models.JobJdbcDatasource.id == param.writer_datasource_id).first()
    if not writer_ds:
        raise Exception('Writer datasource not found')

    engine = DataXEngine()
    job = engine.build_job(reader_ds, writer_ds, param, is_show=False)
    logger.info(job)
    return {'json': job}