datax-admin/app/routers/job_jdbc_datasource.py

from typing import Optional
from fastapi import APIRouter

from fastapi import Depends
from sqlalchemy.orm import Session
from app import schemas

import app.crud as crud
from utils.sx_time import sxtimeit
from utils.sx_web import web_try
from fastapi_pagination import Page, add_pagination, paginate, Params

from app import get_db



router = APIRouter(
    prefix="/jpt/datasource",
    tags=["datasource-数据源管理"],
)


@router.post("/test")
@web_try()
@sxtimeit
def test_datasource_connection(ds: schemas.JobJdbcDatasourceCreate, db: Session = Depends(get_db)):
    return crud.test_datasource_connection(db, ds)


@router.post("/preview")
@web_try()
@sxtimeit
def get_preview_data(ds_id: int, table_name: str, limit: int = 100, db: Session = Depends(get_db)):
    return crud.get_preview_data(db, ds_id, table_name, limit)


@router.post("/table_names", description="获取所有表名")
@web_try()
@sxtimeit
def get_table_names(ds_id: int, db: Session = Depends(get_db)):
    return crud.get_table_names(db, ds_id)

@router.post("/table_schema", description="获取表结构信息")
@web_try()
@sxtimeit
def get_table_schema(ds_id: int, table_name: str, db: Session = Depends(get_db)):
    return crud.get_table_schema(db, ds_id, table_name)

@router.post("/")
@web_try()
@sxtimeit
def create_datasource(ds: schemas.JobJdbcDatasourceCreate, db: Session = Depends(get_db)):
    return crud.create_job_jdbc_datasource(db, ds)





@router.get("/")
@web_try()
@sxtimeit
def get_datasources(datasource_type: Optional[str] = None, params: Params=Depends(), db: Session = Depends(get_db)):
    return paginate(crud.get_job_jdbc_datasources(db, datasource_type), params)

@router.get("/info")
@web_try()
@sxtimeit
def get_datasources_info(ds_id: int, db: Session = Depends(get_db)):
    return crud.get_job_jdbc_datasources_info(db, ds_id)

@router.put("/{ds_id}")
@web_try()
@sxtimeit
def update_datasource(ds_id: int, ds: schemas.JobJdbcDatasourceUpdate, db: Session = Depends(get_db)):
    return crud.update_job_jdbc_datasources(db, ds_id, ds)

@router.delete("/{ds_id}")
@web_try()
@sxtimeit
def delete_job_jdbc_datasource(ds_id: int, db: Session = Depends(get_db)):
    return crud.delete_job_jdbc_datasource(db, ds_id)

add_pagination(router)