01_Databricks测试

import oracledb

try:
    dsn = f"{host}:{port}/{service_name}"
    connection = oracledb.connect(user=username, password=password, dsn=dsn)
    cursor = connection.cursor()

    file_id = 1
    cursor.execute("SELECT FILE_DATA FROM C##TMT.FILE_STORAGE WHERE file_id = :1", [file_id])
    result = cursor.fetchone()

    if result:
        file_data = result[0]
        with open("D:\workspace\GitHub\Databricks2.7z", "wb") as f:
            f.write(file_data.read())
        print(f"文件ID {file_id} 下载成功")
    else:
        print(f"未找到文件ID {file_id}")

    print("\n" + "="*50)

except Exception as e:
    print(f"错误: {e}")

finally:
    if 'cursor' in locals():
        cursor.close()
    if 'connection' in locals():
        connection.close()