DataOps-platform/test_check_data_products.py

#!/usr/bin/env python3
"""检查数据产品表记录"""
from app import create_app, db
from sqlalchemy import text

app = create_app()
with app.app_context():
    # 查询最新的数据产品记录
    result = db.session.execute(text('''
        SELECT id, product_name, product_name_en, target_table, source_dataflow_name, created_at 
        FROM data_products 
        ORDER BY created_at DESC 
        LIMIT 5
    ''')).fetchall()
    
    print('最新的数据产品记录:')
    print('-' * 100)
    for row in result:
        print(f'ID: {row[0]}')
        print(f'  产品名: {row[1]}')
        print(f'  英文名: {row[2]}')
        print(f'  目标表: {row[3]}')
        print(f'  来源数据流: {row[4]}')
        print(f'  创建时间: {row[5]}')
        print('-' * 100)