合并分支

app/routers/dag.py

@@ -21,7 +21,7 @@ router = APIRouter(
 @router.post("/execute")
 @web_try()
 @sxtimeit
-def execute_dag(dag: schemas.Dag,db: Session = Depends(get_db)):
+def execute_dag(dag: schemas.Dag, db: Session = Depends(get_db)):
     af_job = dag_job_submit(dag.dag_uuid, dag.dag_script,db)
     return af_job
 

configs/settings.py

@@ -25,6 +25,8 @@ elif os.environ.get('APP_ENV') == 'production':
     config.readfp(open('production.ini'))
 elif os.environ.get('APP_ENV') == 'idctest':
     config.readfp(open('idctest.ini'))
+elif os.environ.get('APP_ENV') == 'sxkj':
+    config.readfp(open('sxkj.ini'))
 
 print(f"get config of {os.environ.get('APP_ENV')}")
 print(config.get('DATABASE', 'host'))

data/data.sql

@@ -345,17 +345,17 @@ ADD COLUMN `status` tinyint(4) NOT NULL COMMENT '状态(1:转存中,2:成功,3:
 -- ----------------------------
 -- Table structure for project
 -- ----------------------------
+DROP TABLE IF EXISTS `project`;
 CREATE TABLE `project` (
   `id` bigint(20) NOT NULL AUTO_INCREMENT,
   `name` varchar(30) COLLATE utf8_unicode_ci NOT NULL COMMENT '项目名称',
-  `code` varchar(30) COLLATE utf8_unicode_ci NOT NULL COMMENT '项目编码',
+  `code` varchar(128) COLLATE utf8_unicode_ci NOT NULL COMMENT '项目编码',
   `type` tinyint(4) NOT NULL COMMENT '项目类型（0:算法组；1:普通项目）',
   `create_time` int(20) NOT NULL COMMENT '创建时间',
   `project_token` varchar(1000) COLLATE utf8_unicode_ci DEFAULT NULL COMMENT 'project_token',
   PRIMARY KEY (`id`)
 ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='项目';
 
-
 -- ----------------------------
 -- Records of project
 -- ----------------------------
@@ -366,10 +366,11 @@ COMMIT;
 -- ----------------------------
 -- Table structure for users
 -- ----------------------------
+DROP TABLE IF EXISTS `users`;
 CREATE TABLE `users` (
   `id` bigint(20) NOT NULL AUTO_INCREMENT,
   `name` varchar(30) COLLATE utf8_unicode_ci NOT NULL COMMENT '姓名',
-  `code` varchar(30) COLLATE utf8_unicode_ci NOT NULL COMMENT '用户编码',
+  `code` varchar(128) COLLATE utf8_unicode_ci NOT NULL COMMENT '用户编码',
   `username` varchar(30) COLLATE utf8_unicode_ci NOT NULL COMMENT '用户名',
   `password` varchar(128) COLLATE utf8_unicode_ci NOT NULL COMMENT '密码暗文',
   `create_time` int(20) NOT NULL COMMENT '创建时间',
@@ -387,10 +388,11 @@ COMMIT;
 -- ----------------------------
 -- Table structure for roles
 -- ----------------------------
+DROP TABLE IF EXISTS `roles`;
 CREATE TABLE `roles` (
   `id` bigint(20) NOT NULL,
-  `name` varchar(50) COLLATE utf8_unicode_ci NOT NULL COMMENT '角色名',
-  `code` varchar(50) COLLATE utf8_unicode_ci NOT NULL COMMENT '角色编码',
+  `name` varchar(30) COLLATE utf8_unicode_ci NOT NULL COMMENT '角色名',
+  `code` varchar(128) COLLATE utf8_unicode_ci NOT NULL COMMENT '角色编码',
   PRIMARY KEY (`id`)
 ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='角色';
 
@@ -408,6 +410,7 @@ COMMIT;
 -- ----------------------------
 -- Table structure for project_user_relation
 -- ----------------------------
+DROP TABLE IF EXISTS `project_user_relation`;
 CREATE TABLE `project_user_relation` (
   `id` bigint(20) NOT NULL,
   `user_id` bigint(20) NOT NULL COMMENT '用户id',
@@ -426,6 +429,7 @@ COMMIT;
 -- ----------------------------
 -- Table structure for data_table
 -- ----------------------------
+DROP TABLE IF EXISTS `data_table`;
 CREATE TABLE `data_table` (
   `id` bigint(20) NOT NULL AUTO_INCREMENT,
   `database_name` varchar(100) COLLATE utf8_unicode_ci DEFAULT NULL COMMENT '库名',

docker-compose.yml

@@ -20,3 +20,4 @@ services:
       - 'minio-api.sxkj.com:192.168.199.109'
     environment:
       - APP_ENV=development
+      - APP_PREFIX=jpt

server.py

@@ -22,10 +22,15 @@ from app.routers.job import router_af_job
 from app.routers.task import router_af_task
 from configs.globals import GlobalsMiddleware
 from utils.sx_log import format_print
+import os
 
 format_print()
+PREFIX = os.environ.get('APP_PREFIX', default='jpt')
+print(f'prefix: {PREFIX}')
+
+
 Base.metadata.create_all(bind=engine)
-app = FastAPI(docs_url='/jpt/docs', redoc_url='/jpt/redoc', title="DAG In Airflow管理系统")
+app = FastAPI(docs_url=f'/{PREFIX}/docs', redoc_url=f'/{PREFIX}/redoc', openapi_url=f'/{PREFIX}/openapi.json', title="DAG In Airflow管理系统")
 
 
 # CORS 跨源资源共享
@@ -57,13 +62,14 @@ app.include_router(router_jm_job_info.router)
 app.include_router(router_jm_job_log.router)
 app.include_router(router_code_check.router)
 
-# AirflowJobSubmitter.auto_submit_data_transfer()
-# AirflowJobSubmitter.auto_submit_requirements_install()
+if PREFIX == 'af':
+    AirflowJobSubmitter.auto_submit_data_transfer()
+    AirflowJobSubmitter.auto_submit_requirements_install()
 print('server init finish:)!!!')
 
 
 # Get 健康检查
-@app.get("/jpt/ping", description="健康检查")
+@app.get(f"/{PREFIX}/ping", description="健康检查")
 def ping():
     return "pong!!"
 

sxkj.ini

@@ -55,7 +55,6 @@ sparks=SXKJ:32775/jupyter:0.981
 
 
 [HADOOP_INNER]
-datasource = 1,2,3
 default_fs = hdfs://192.168.199.27:9000
 hadoop_config={
              }
@@ -64,6 +63,17 @@ kerberos_config = {
                     "kerberosKeytabFilePath": "/workspace/confs/test/user.keytab",
                     "kerberosPrincipal": "ailab@EMR-5XJSY31F"
                   }
+
+[LAKE_HADOOP_INNER]
+default_fs = hdfs://192.168.199.27:9000
+hadoop_config={
+             }
+kerberos_config = {
+                    "haveKerberos": "false",
+                    "kerberosKeytabFilePath": "/workspace/confs/test/user.keytab",
+                    "kerberosPrincipal": "ailab@EMR-5XJSY31F"
+                  }
+
 [HOST_ALIAS]
 enable = true
 host_alias = {