datax-admin/app/routers/code_check.py

import json
import os
import tempfile
from typing import Optional
from fastapi import APIRouter
from fastapi import Depends
from app import schemas
from app.common.decorators import verify_all
from utils.sx_time import sxtimeit
from utils.sx_web import web_try
from pylint import epylint
import sqlfluff


router = APIRouter(
    prefix="/jpt/code_check",
    tags=["code_check-代码校验"],
)

@router.post("/", dependencies=[Depends(verify_all)])
@web_try()
@sxtimeit
def code_check(code: str, code_type: str):
    res = []
    if code_type == "python":
        file = tempfile.NamedTemporaryFile(delete=False, mode='w')
        file.write(code)
        file.close()

        options = ' '.join([
            file.name,
            '--output-format', 'json',
        ])

        (lint_stdout, lint_stderr) = epylint.py_run(return_std=True, command_options=options)

        os.remove(file.name)
        res = json.loads(lint_stdout.getvalue())
    else:
        lint_out = sqlfluff.lint(code,'ansi')
        res = [l for l in lint_out if l['code'] == 'PRS']
    return res