数据源、同步配置重名验证

app/crud/job_jdbc_datasource.py

@@ -72,8 +72,10 @@ def create_job_jdbc_datasource(db: Session, item: schemas.JobJdbcDatasourceCreat
         item.jdbc_password = decode_base64(item.jdbc_password)
     ds, item = _format_datasource(db, item)
 
-    #
     create_time: int = int(time.time())
+    name_item = db.query(models.JobJdbcDatasource).filter(models.JobJdbcDatasource.datasource_name == item.datasource_name).first()
+    if name_item:
+        raise Exception('数据源名称重复')
     db_item = models.JobJdbcDatasource(**item.dict(), **{
         'status': 1,
         'create_time': create_time,
@@ -120,6 +122,11 @@ def update_job_jdbc_datasources(db: Session, ds_id: int, update_item: schemas.Jo
     db_item = db.query(models.JobJdbcDatasource).filter(models.JobJdbcDatasource.id == ds_id).first()
     if not db_item:
         raise Exception('未找到该数据源')
+    name_item = db.query(models.JobJdbcDatasource)\
+        .filter(models.JobJdbcDatasource.datasource_name == update_item.datasource_name)\
+        .filter(models.JobJdbcDatasource.id != ds_id).first()
+    if name_item:
+        raise Exception('数据源名称重复')
     update_dict = update_item.dict(exclude_unset=True)
     for k, v in update_dict.items():
         setattr(db_item, k, v)

app/services/job_info.py

@@ -8,6 +8,10 @@ import app.crud as crud
 def create_job_info_services(db: Session, item: schemas.JobInfoCreate):
     create_time: int = int(time.time())
     item_dict = item.dict()
+    name_item = db.query(models.JobInfo)\
+        .filter(models.JobInfo.job_desc == item.job_desc).first()
+    if name_item:
+        raise Exception('同步配置名称重复')
     # 定时任务对象转为cron表达式
     cron_expression_dict = item_dict.pop('cron_expression')
     cron_expression = joint_cron_expression(schemas.CronExpression(**cron_expression_dict))
@@ -46,6 +50,11 @@ def create_job_info_services(db: Session, item: schemas.JobInfoCreate):
 def update_job_info_services(db: Session, id: int, update_item: schemas.JobInfoUpdate):
     # 获取任务信息
     db_item = crud.get_job_info(db,id)
+    name_item = db.query(models.JobInfo)\
+        .filter(models.JobInfo.job_desc == update_item.job_desc)\
+        .filter(models.JobInfo.id != id).first()
+    if name_item:
+        raise Exception('同步配置名称重复')
     update_dict = update_item.dict(exclude_unset=True)
     # 定时任务对象转为cron表达式
     cron_expression_dict = update_dict.pop('cron_expression')