使用Python Great Expectations和pythonoracledb

这为我创建了一个数据源:

import getpass
import os
import oracledb
import great_expectations as gx

un = os.environ.get('PYTHON_USERNAME')
cs = os.environ.get('PYTHON_CONNECTSTRING')
pw = getpass.getpass(f'Enter password for {un}@{cs}: ')

# I use Easy Connect strings like "localhost/orclpdb1".  These two lines
# let me access the components individually
cp = oracledb.ConnectParams()
cp.parse_connect_string(cs)

connection_string = f'oracle+oracledb://{un}:{pw}@{cp.host}:{cp.port}/?service_name={cp.service_name}'

context = gx.get_context()

datasource = context.sources.add_sql(
    name="my_datasource", connection_string=connection_string
)

asset_name = "my_query_asset"
query = "select * from employees"

query_asset = datasource.add_query_asset(name=asset_name, query=query)

for i in query_asset:
    print(i)

运行它会得到:

('name', 'my_query_asset')
('type', 'query')
('id', None)
('order_by', [])
('batch_metadata', {})
('splitter', None)
('query', 'select * from employees')

有关将oracledb与SQLAlchemy连接的信息，请参阅Using SQLAlchemy 2.0 with python-oracledb for Oracle Database.