2021.8.10 14:44 update table_refresh with task

nheweijun
app/modules/manager/table_refresh.py
--- a/app/modules/manager/table_refresh.py
查看文件 @205d293
+++ b/app/modules/manager/table_refresh.py
查看文件 @205d293
@@ -5,7 +5,7 @@
 import traceback
-from app.models import Table,Database,DES,Columns,db,TableVacuate
+from app.models import Table,Database,DES,Columns,db,TableVacuate,Task,Process
 from osgeo.ogr import DataSource,FeatureDefn,FieldDefn,Layer
@@ -20,59 +20,108 @@ from app.util.component.StructuredPrint import StructurePrint
 from app.util.component.ApiTemplate import ApiTemplate
 from app.util.component.GeometryAdapter import GeometryAdapter
+import multiprocessing
+import configure
+
 class Api(ApiTemplate):
     api_name = "数据刷新"
     def process(self):
-
-
         res = {}
-        pg_ds =None
-        data_session=None
-        this_time = datetime.datetime.now()
         try:
-            
+
             database_guid = self.para.get("database_guid")
             database = Database.query.filter_by(guid=database_guid).one_or_none()
-    
+
             if not database:
                 raise Exception("数据库不存在!")
-            spatial_tables = Table.query.order_by(Table.create_time.desc()).filter_by(database_guid=database_guid).filter(Table.table_type!=0).all()
+                # 初始化task
+            task_guid = uuid.uuid1().__str__()
+
+            # result = self.table_refresh(database)
+
+            refresh_process = multiprocessing.Process(target=self.table_refresh,args=(database,task_guid))
+            refresh_process.start()
+
+            task = Task(guid=task_guid,
+                        name="数据库{}更新".format(database.alias),
+                        create_time=datetime.datetime.now(),
+                        state=0,
+                        task_type=3,
+                        creator=self.para.get("creator"),
+                        file_name=None,
+                        database_guid=database.guid,
+                        process="数据库更新中")
+
+            db.session.add(task)
+            db.session.commit()
+            res["msg"] = "数据库更新已提交！"
+            res["data"] = task_guid
+            res["result"] = True
+
+        except Exception as e:
+            print(traceback.format_exc())
+            raise e
+        return res
+
+
+    def table_refresh(self,database,task_guid):
+
+        pg_ds =None
+        data_session=None
+        result = {}
+        sys_session = None
+        try:
+            sys_session = PGUtil.get_db_session(configure.SQLALCHEMY_DATABASE_URI)
+
+
+            this_time = datetime.datetime.now()
+            database_guid = database.guid
+
+            # 已注册空间表
+            spatial_tables = sys_session.query(Table).order_by(Table.create_time.desc()).filter_by(database_guid=database_guid).filter(
+                Table.table_type != 0).all()
+
+            # 已注册空间表名
             spatial_tables_names = [table.name for table in spatial_tables]
-            db_tables_names = []
-    
+            # 实体库datasource
             pg_ds: DataSource = PGUtil.open_pg_data_source(1, DES.decode(database.sqlalchemy_uri))
+
             # 更新空间表
             # 增加表
-            self.add_spatail_table(database, pg_ds, spatial_tables_names, db_tables_names, this_time)
-    
+            db_tables_names = self.add_spatail_table(database, pg_ds, sys_session,spatial_tables_names, this_time)# 实体库中空间表名
+
             # 删除/修改表
-            self.edit_spatial_table(pg_ds, spatial_tables, db_tables_names, this_time)
+            self.edit_spatial_table(pg_ds, sys_session,spatial_tables, db_tables_names, this_time)
-            # 处理抽稀表
-            self.deal_vacuate_table(pg_ds,database.guid)
+            # 空间表处理完毕
+            sys_session.commit()
+            # 处理抽稀表
+            self.deal_vacuate_table(pg_ds,sys_session, database.guid)
             # 空间表处理完毕
-            db.session.commit()
+            sys_session.commit()
-    
             # 注册普通表
             # 实体库连接
             data_session: Session = PGUtil.get_db_session(DES.decode(database.sqlalchemy_uri))
-    
-            # 空间表
-            spatial_tables = Table.query.order_by(Table.create_time.desc()).filter_by(database_guid=database_guid).filter(Table.table_type!=0).all()
+
+            # 处理后空间表
+            spatial_tables = sys_session.query(Table).order_by(Table.create_time.desc()).filter_by(database_guid=database_guid).filter(
+                Table.table_type != 0).all()
+            # 处理后空间表名
             spatial_tables_names = [table.name for table in spatial_tables]
-    
-    
-            #原有普通表
-            common_tables = Table.query.order_by(Table.create_time.desc()).filter_by(database_guid=database_guid).filter(Table.table_type==0).all()
+
+            # 原有普通表
+            common_tables = sys_session.query(Table).order_by(Table.create_time.desc()).filter_by(database_guid=database_guid).filter(
+                Table.table_type == 0).all()
+            # 原有普通表 名
             origin_common_tables_name = [table.name for table in common_tables]
-    
-    
+
             # 现有普通表
-            real_common_tables_name =[]
+            real_common_tables_name = []
+
             # 只注册public中的表
             common_result = data_session.execute(
                 "select relname as tabname from pg_class c where  relkind = 'r' and relnamespace=2200 and  relname not like 'pg_%' and relname not like 'sql_%' order by relname").fetchall()
@@ -80,36 +129,56 @@ class Api(ApiTemplate):
                 table_name = re[0]
                 if table_name not in spatial_tables_names and (not table_name.__contains__("_vacuate_")):
                     real_common_tables_name.append(table_name)
-    
+
             # 增加新普通表
-            self.add_common_table(data_session, database_guid, real_common_tables_name, origin_common_tables_name,
-                            this_time)
+            self.add_common_table(data_session, sys_session, database_guid, real_common_tables_name, origin_common_tables_name,
+                                  this_time)
-    
-            #删除、修改普通表
-            self.edit_common_table(data_session, database_guid, real_common_tables_name, origin_common_tables_name,
-                              this_time)
+            # 删除、修改普通表
+            self.edit_common_table(data_session,sys_session, database_guid, real_common_tables_name, origin_common_tables_name,
+                                   this_time)
-            db.session.commit()
-            res["msg"] = "刷新数据成功！"
-            res["result"] = True
+            sys_session.commit()
+            result["data"] = "刷新数据成功！"
+            result["state"] = 1
+            sys_session.query(Task).filter_by(guid=task_guid).update(
+                {"state": 1, "update_time": datetime.datetime.now(),"process":"更新成功"})
+            sys_session.commit()
         except Exception as e:
-            raise e
-    
+            try:
+                print(traceback.format_exc())
+                sys_session.query(Task).filter_by(guid=task_guid).update(
+                    {"state": -1, "update_time": datetime.datetime.now(),"process":"更新失败"})
+                message = "{} {}".format(datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S'), e.__str__())
+                task_process_guid = uuid.uuid1().__str__()
+                task_process = Process(guid=task_process_guid, message=message, time=datetime.datetime.now(),
+                                       task_guid=task_guid)
+                sys_session.add(task_process)
+                sys_session.commit()
+            except Exception as ee:
+                print(traceback.format_exc())
         finally:
             if pg_ds:
                 pg_ds.Destroy()
             if data_session:
                 data_session.close()
-    
-        return res
-    
-
-    def add_spatail_table(self,database,pg_ds,spatial_tables_names,db_tables_names,this_time):
-        # 更新空间表
-        # 增加表
+            if sys_session:
+                sys_session.close()
+        return result
+
+    def add_spatail_table(self,database,pg_ds,sys_session,spatial_tables_names,this_time):
+        '''
+        注册新增空间表
+        :param database:
+        :param pg_ds:
+        :param spatial_tables_names: 已注册空间表名
+        :param this_time:
+        :return: 实体库中空间表名
+        '''
+
+        db_tables_names=[]
         for i in range(pg_ds.GetLayerCount()):
             layer: Layer = pg_ds.GetLayer(i)
@@ -123,7 +192,7 @@ class Api(ApiTemplate):
                 # 只注册public的空间表,其他表空间的表名会有.
                 if layer.GetName().__contains__("."):
                     continue
-
+                # 略过抽稀表
                 if layer.GetName().__contains__("_vacuate_"):
                     continue
@@ -152,14 +221,13 @@ class Api(ApiTemplate):
                 table_guid = uuid.uuid1().__str__()
                 table = Table(guid=table_guid,
                               database_guid=database.guid,
-                              alias=layer.GetName(),
+                              # alias=layer.GetName(),
                               name=layer.GetName(), create_time=this_time, update_time=this_time,
                               table_type=GeometryAdapter.get_table_type(layer.GetGeomType()),
                               extent=extent,
                               feature_count=feature_count
                               )
-                db.session.add(table)
-
+                sys_session.add(table)
                 feature_defn: FeatureDefn = layer.GetLayerDefn()
                 for i in range(feature_defn.GetFieldCount()):
@@ -169,9 +237,10 @@ class Api(ApiTemplate):
                         "") else field_defn.GetAlternativeName()
                     column = Columns(guid=uuid.uuid1().__str__(), table_guid=table_guid,
                                      name=field_name, alias=field_alias, create_time=this_time, update_time=this_time)
-                    db.session.add(column)
+                    sys_session.add(column)
+        return db_tables_names
-    def deal_vacuate_table(self,pg_ds,database_guid):
+    def deal_vacuate_table(self,pg_ds,sys_session,database_guid):
         for i in range(pg_ds.GetLayerCount()):
@@ -198,30 +267,30 @@ class Api(ApiTemplate):
                 except:
                     pass
-                base_table =Table.query.filter_by(name=base_layer_name,database_guid=database_guid).one_or_none()
+                base_table =sys_session.query(Table).filter_by(name=base_layer_name,database_guid=database_guid).one_or_none()
                 if base_table:
-                    if not TableVacuate.query.filter_by(table_guid=base_table.guid,name=l_name).one_or_none():
+                    if not sys_session.query(TableVacuate).filter_by(table_guid=base_table.guid,name=l_name).one_or_none():
                         table_vacuate = TableVacuate(guid=uuid.uuid1().__str__(),
                                                      table_guid=base_table.guid,
                                                      level=level,
                                                      name=l_name,
                                                      pixel_distance=float(pixel_distance_str))
-                        db.session.add(table_vacuate)
-                        Table.query.filter_by(guid=base_table.guid).update({"is_vacuate": 1})
+                        sys_session.add(table_vacuate)
+
+                        sys_session.query(Table).filter_by(guid=base_table.guid).update({"is_vacuate": 1})
                 else:
                     kk=1
-
-    def edit_spatial_table(self,pg_ds,spatial_tables,db_tables_names,this_time):
+    def edit_spatial_table(self,pg_ds,sys_session,spatial_tables,db_tables_names,this_time):
         for table in spatial_tables:
             # 删除表
             if table.name not in db_tables_names:
                 StructurePrint.print("空间表减少！")
-                db.session.delete(table)
+                sys_session.delete(table)
             # 修改表
             else:
                 layer: Layer = pg_ds.GetLayerByName(table.name)
@@ -252,15 +321,13 @@ class Api(ApiTemplate):
                         column = Columns(guid=uuid.uuid1().__str__(), table_guid=table.guid,
                                          name=field_name, alias=field_alias, create_time=this_time,
                                          update_time=this_time)
-                        db.session.add(column)
+                        sys_session.add(column)
                 # 删除列
                 for column in columns:
                     if column.name not in db_columns_names:
                         StructurePrint.print("{}空间表属性减少！".format(table.name))
-                        db.session.delete(column)
-
-
+                        sys_session.delete(column)
                 # 范围统计和数量统计以100w为界限
                 query_count_layer: Layer = pg_ds.ExecuteSQL(
@@ -271,6 +338,7 @@ class Api(ApiTemplate):
                 if feature_count < 1000000:
                     feature_count = layer.GetFeatureCount()
                     ext = layer.GetExtent()
+
                 else:
                     query_ext_layer: Layer = pg_ds.ExecuteSQL(
                         "select geometry(ST_EstimatedExtent('public', '{}','{}'))".format(l_name,
@@ -285,11 +353,11 @@ class Api(ApiTemplate):
                 # 修改要素量
                 if not table.feature_count.__eq__(feature_count):
                     StructurePrint.print("{}空间表要素！".format(table.name))
-                    Table.query.filter_by(guid=table.guid).update({"feature_count": feature_count,
+                    sys_session.query(Table).filter_by(guid=table.guid).update({"feature_count": feature_count,
                                                                    "extent": extent})
-    def add_common_table(self,data_session,database_guid,real_common_tables_name,origin_common_tables_name,this_time):
+    def add_common_table(self,data_session,sys_session,database_guid,real_common_tables_name,origin_common_tables_name,this_time):
         for table_name in real_common_tables_name:
             if table_name not in origin_common_tables_name:
                 StructurePrint.print("{}非空间表增加！".format(table_name))
@@ -303,7 +371,7 @@ class Api(ApiTemplate):
                               feature_count=count
                               )
-                db.session.add(table)
+                sys_session.add(table)
                 sql = '''
                    SELECT
@@ -325,22 +393,22 @@ class Api(ApiTemplate):
                 for col in cols:
                     column = Columns(guid=uuid.uuid1().__str__(), table_guid=table_guid,
                                      name=col[1], create_time=this_time, update_time=this_time)
-                    db.session.add(column)
+                    sys_session.add(column)
                 # 删除不存在的表
                 for n in origin_common_tables_name:
                     if n not in real_common_tables_name:
                         tables = Table.query.filter_by(name=n).filter_by(database_guid=database_guid).all()
                         for table in tables:
-                            db.session.delete(table)
+                            sys_session.delete(table)
-    def edit_common_table(self,data_session,database_guid,real_common_tables_name,origin_common_tables_name,this_time):
+    def edit_common_table(self,data_session,sys_session,database_guid,real_common_tables_name,origin_common_tables_name,this_time):
         for table_name in origin_common_tables_name:
-            tables = Table.query.filter_by(name=table_name).filter_by(database_guid=database_guid).all()
+            tables = sys_session.query(Table).filter_by(name=table_name).filter_by(database_guid=database_guid).all()
             for table in tables:
                 if table_name not in real_common_tables_name:
                     StructurePrint.print("{}非空间表减少！".format(table_name))
-                    db.session.delete(table)
+                    sys_session.delete(table)
                 # 修改表
                 else:
                     columns = table.relate_columns
@@ -371,13 +439,13 @@ class Api(ApiTemplate):
                             StructurePrint.print("{}表要素属性增加！".format(table_name))
                             column = Columns(guid=uuid.uuid1().__str__(), table_guid=table.guid,
                                              name=col, create_time=this_time, update_time=this_time)
-                            db.session.add(column)
+                            sys_session.add(column)
                     # 属性减少
                     for column in columns:
                         if column.name not in real_cols_name:
                             StructurePrint.print("{}表要素属性减少！".format(table_name))
-                            db.session.delete(column)
+                            sys_session.delete(column)
                     # 修改要素量
                     sql = 'select count(*) from "{}"'.format(table_name)
@@ -385,7 +453,7 @@ class Api(ApiTemplate):
                     count = data_session.execute(sql).fetchone()[0]
                     if not table.feature_count.__eq__(count):
                         StructurePrint.print("{}表要素变化！".format(table_name))
-                        Table.query.filter_by(guid=table.guid).update({"feature_count": count})
+                        sys_session.query(Table).filter_by(guid=table.guid).update({"feature_count": count})