Save table metadata to database
(table_name, table_metadata)
| 16 | |
| 17 | |
| 18 | def save_table_metadata(table_name, table_metadata): |
| 19 | """ |
| 20 | Save table metadata to database |
| 21 | """ |
| 22 | if not DB_MANAGED_METADATA: |
| 23 | TABLES_METADATA_DICT[table_name] = table_metadata |
| 24 | save_tables_metadata_to_json() |
| 25 | return |
| 26 | try: |
| 27 | tm = TableMetadata.query.filter_by(table_name=table_name).one_or_none() |
| 28 | if tm: |
| 29 | tm.table_metadata = table_metadata |
| 30 | else: |
| 31 | tm = TableMetadata( |
| 32 | table_name=table_name, |
| 33 | table_metadata=table_metadata |
| 34 | ) |
| 35 | db.session.add(tm) |
| 36 | db.session.commit() |
| 37 | except Exception as e: |
| 38 | print(e) |
| 39 | db.session.rollback() |
| 40 | |
| 41 | |
| 42 | def save_type_metadata(type_name, type_metadata): |
no test coverage detected