Usage¶

This section covers basic usage of the IQVIA NLP Airflow provider, demonstrating how to use Connections, Operators, Transfer operators, and Hooks. For detailed reference information about any of the objects mentioned here, refer to the Python API section.

Connections¶

The I2E connection type¶

The I2E connection type is used to connect to an I2E server from Airflow. To create a new connection from the Airflow web interface, select the Admin tab, open the Connections page, and create a new connection of type I2E. The I2E connection type has the following fields:

Connection Id	The connection ID. This must be unique across the Airflow installation.
Connection Type	The connection type. This should be set to `I2E`.
Description	A human-readable description for the connection.
Host	The hostname of the I2E server. This shouldn’t include the scheme or port.
Scheme	The scheme of the connection, e.g. `http` or `https`. Defaults to `https`.
Username	The username for the I2E service account that is used to authenticate with the I2E server.
Password	The password for the I2E service account that is used to authenticate with the I2E server.
Port	The port number of the I2E server. Defaults to `80` for `http` and `443` for `https`.
Verify SSL	Whether to verify the SSL certificate of the I2E server.
License Pool	The I2E license pool to use when making requests to the I2E server.

Once the connection has been created, you can pass the connection ID to various operators in your DAGs to connect to that I2E server. For example, if you created a connection with the ID i2e_default, then you would pass it to the I2ERunQueryOperator like so:

from iqvia_nlp_provider.operators.i2e import I2ERunQueryOperator

I2ERunQueryOperator(i2e_conn_id="i2e_default", ...)

If you have multiple I2E servers, you should create a connection for each of them. The I2EPickConnectionIDOperator is used to load balance between multiple I2E servers.

Operators¶

I2EMakeIndexOperator¶

The I2EMakeIndexOperator builds an index in I2E, and returns the URI of the index. This example shows how to use the I2EMakeIndexOperator to index the source data file source.txt:

from iqvia_nlp_provider.operators.i2e import I2EMakeIndexOperator

i2e_index_uri = I2EMakeIndexOperator(
    i2e_conn_id="i2e_default",
    i2e_source_data_uri="/api;type=source_data/source.txt",
    i2e_index_template="data_factory/sdoh-2_0_0/sdoh",
)

This example assumes that the source data file source.txt is already present in I2E. In practice, you would likely use a transfer operator (S3ToI2EOperator, SQLTimestampToI2EOperator, or LocalFilesystemToI2EOperator) to transfer source data from an external datastore to I2E. The return value from the transfer operator would provide the i2e_source_data_uri value for the I2EMakeIndexOperator.

I2ERunQueryOperator¶

The I2ERunQueryOperator runs a query in I2E, and returns the URI of the query results. This example shows how to use the I2ERunQueryOperator to run a query against the index data_factory/sdoh-2_0_0/sdoh:

from iqvia_nlp_provider.operators.i2e import I2ERunQueryOperator

i2e_query_results_uri = I2ERunQueryOperator(
    i2e_conn_id="i2e_default",
    i2e_index_uri="/api;type=index/sdoh-index",
    i2e_query_template="data_factory/sdoh-2_1_0/All Social Determinants of Health",
)

Typically you would use this operator after the I2EMakeIndexOperator, using the return value from the I2EMakeIndexOperator as the i2e_index_uri value for the I2ERunQueryOperator. After using the I2ERunQueryOperator the query results will be present in I2E, ready for them to be retrieved or copied to an external datastore.

I2EPrintTaskLogOperator¶

The I2EPrintTaskLogOperator prints the task log for a given task URI. This example shows how to use the I2EPrintTaskLogOperator to print the task log for a given task URI:

from iqvia_nlp_provider.operators.i2e import I2EPrintTaskLogOperator

I2EPrintTaskLogOperator(i2e_conn_id="i2e_default")

Typically you would use this operator after the I2ERunQueryOperator, or I2EMakeIndexOperator. These classes set the task_uri XCom value, which is then passed to the I2EPrintTaskLogOperator. You can also specify a custom task URI using the task_uri argument, e.g. task_uri="/api;type=query_task/task_id".

I2EPickConnectionIDOperator¶

The I2EPickConnectionIDOperator is used to load balance between multiple I2E servers by picking an I2E connection ID from a given list, according to the given Strategy:

Strategy.RANDOM - Randomly selects one of the connection IDs.
Strategy.ROUND_ROBIN - Cycles through the connection IDs in order.
Strategy.LEAST_BUSY - Selects the connection ID with the fewest running tasks.

The connection ID returned by this operator can then be passed to other operators in your DAG.

In Airflow 3, the provider uses the Airflow REST API v2 to list I2E connections and to read previous DAG run XComs for the round-robin strategy. Airflow workers must therefore be able to reach the Airflow API server and obtain an auth token from /auth/token. In Docker or Kubernetes deployments, make sure the API server base URL is reachable from workers; for example, the development Docker Compose stack sets AIRFLOW__API__BASE_URL and AIRFLOW__CORE__EXECUTION_API_SERVER_URL to the internal API server URL.

This example shows how to use the I2EPickConnectionIDOperator to pick from the I2E connection IDs i2e_default_1 and i2e_default_2 in a round-robin fashion. Every time this DAG runs, the picked connection ID will alternate between i2e_default_1 and i2e_default_2. The picked connection ID is then passed to the I2EMakeIndexOperator:

from airflow.sdk import dag
from iqvia_nlp_provider.operators.i2e import I2EMakeIndexOperator, I2EPickConnectionIDOperator, Strategy

@dag(schedule=None, catchup=False)
def i2e_query_sdoh():

    i2e_conn_id = I2EPickConnectionIDOperator(
        strategy=Strategy.ROUND_ROBIN,
        choices=["i2e_default_1", "i2e_default_2"],
    )

    I2EMakeIndexOperator(i2e_conn_id=i2e_conn_id.output, ...)

i2e_query_sdoh()

I2EDeleteResourcesOperator¶

The I2EDeleteResourcesOperator deletes one or more resources from I2E. This example shows how to delete a source data file and an index from I2E:

from iqvia_nlp_provider.operators.i2e import I2EDeleteResourcesOperator

I2EDeleteResourcesOperator(
    i2e_conn_id="i2e_default",
    i2e_resource_uris=["/api;type=source_data/source.txt", "/api;type=index/sdoh-index"],
)

Typically you would use this operator at the end of a DAG to delete any transient I2E resources created during the DAG run. Values for i2e_resource_uris would come from the return values of other operators.

The I2EDeleteResourcesOperator is also used to delete all I2E resources in a particular folder. This example shows how to delete all source data files in the I2E folder sdoh_query_resources:

from iqvia_nlp_provider.operators.i2e import I2EDeleteResourcesOperator

I2EDeleteResourcesOperator(
    i2e_conn_id="i2e_default",
    i2e_resource_types=["source_data"],
    i2e_folder="sdoh_query_resources",
)

Transfer operators¶

Transfer operators are a special type of operator for transferring data between I2E and various datastores. We provide transfer operators for:

Copying source data from an external datastore to I2E.
Copying query results from I2E to an external datastore.
Copying I2E resource bundles from an external datastore to I2E.

Transferring source data¶

These transfer operators only support source document formats that are specified in the I2E documentation, here.

S3ToI2EOperator¶

The S3ToI2EOperator allows a single file or multiple files from an S3 bucket to be uploaded to I2E as a source data file. It returns the URI of the uploaded source data in I2E.

This example shows how to use the S3ToI2EOperator to retrieve a single file named source_s3.txt from the S3 bucket sdoh-query, and upload it to I2E:

from iqvia_nlp_provider.transfers.s3 import S3ToI2EOperator

i2e_source_data_uri = S3ToI2EOperator(
    aws_conn_id="aws_conn_id",
    i2e_conn_id="i2e_default",
    aws_bucket_name="sdoh-query",
    aws_key="source_s3.txt",
)

The operator allows multiple files from an S3 bucket to be uploaded to I2E as a source data file when the select_data_added_between argument is provided. Files are selected from S3 from the given bucket if their creation time falls within a given timespan, and files are zipped into a single file before being uploaded to I2E. This allows you to only process data added since the last DAG run, rather than including all data each time. For example, you can schedule your DAG to run every hour, and set the start date for the S3ToI2EOperator to be one hour before the current time. This way, only the files added to S3 in the last hour are uploaded to I2E.

This example shows how to use the select_data_added_between argument to only select the files added to the S3 bucket sdoh-query within the DAG’s data interval:

from iqvia_nlp_provider.transfers.s3 import S3ToI2EOperator

i2e_source_data_uri = S3ToI2EOperator(
    aws_conn_id="aws_conn_id",
    i2e_conn_id="i2e_default",
    aws_bucket_name="sdoh-query",
    select_data_added_between=("{{ data_interval_start }}", "{{ data_interval_end }}"),
)

Alternatively, you can omit the select_data_added_between argument and provide the name of a prefix for the aws_prefix argument. This will select all files from the S3 bucket that have the given prefix regardless of creation time, and upload them to I2E as source data. You can combine the select_data_added_between argument with the aws_prefix argument to select files with a given prefix that were added within a given timespan.

This example shows how to use the aws_prefix argument to select all files from the given prefix:

from iqvia_nlp_provider.transfers.s3 import S3ToI2EOperator

i2e_source_data_uri = S3ToI2EOperator(
    aws_conn_id="aws_conn_id",
    i2e_conn_id="i2e_default",
    aws_bucket_name="sdoh-query",
    aws_prefix="source_data_files",
)

By omitting both aws_key, aws_prefix, and select_data_added_between, all files from the S3 bucket can be selected:

from iqvia_nlp_provider.transfers.s3 import S3ToI2EOperator

i2e_source_data_uri = S3ToI2EOperator(
    aws_conn_id="aws_conn_id",
    i2e_conn_id="i2e_default",
    aws_bucket_name="sdoh-query",
)

SQLQueryToI2EOperator¶

The SQLQueryToI2EOperator allows data from an SQL database to be uploaded to I2E. It executes a given SQL query and converts the results into a single XML file, which is then uploaded to I2E as a source data file. It returns the URI of the uploaded source data in I2E. The operator is compatible with all database types supported by the apache-airflow-providers-common-sql provider, as listed in the Supported Database Types documentation. The SQLQueryToI2EOperator has been tested against PostgreSQL, MySQL, Microsoft SQL Server, Snowflake, and Databricks.

This example shows how to use the SQLQueryToI2EOperator to select the first 1000 rows from a PostgreSQL table named source_data_table and upload them to I2E:

from iqvia_nlp_provider.transfers.sql import SQLQueryToI2EOperator

i2e_source_data_uri = SQLQueryToI2EOperator(
    sql_conn_id="postgres_default",
    i2e_conn_id="i2e_default",
    sql_query="SELECT * FROM source_data_table LIMIT %(limit)s",
    sql_params={"limit": "1000"}
)

This example shows how to use the SQLQueryToI2EOperator to select a few specific columns (column_1 and column_2) from a PostgreSQL table named source_data_table, where the value of another column (column_3) is within a certain range, and upload the rows to I2E:

from iqvia_nlp_provider.transfers.sql import SQLQueryToI2EOperator

i2e_source_data_uri = SQLQueryToI2EOperator(
    sql_conn_id="postgres_default",
    i2e_conn_id="i2e_default",
    sql_query="SELECT column_1, column_2 FROM source_data_table WHERE column_3 >= %(lower_limit)s AND column_3 < %(upper_limit)s",
    sql_params={"lower_limit": "100", "upper_limit": "200"}
)

SQLTimestampToI2EOperator¶

The SQLTimestampToI2EOperator allows data from an SQL database to be uploaded to I2E. It selects rows created within a given timespan. The creation time of a row is determined by a column in the table that must be populated whenever a new row is added. This feature allows you to only process data added since the last DAG run, rather than including all data each time. For example, you can schedule your DAG to run every hour, and set the start date for the SQLTimestampToI2EOperator to be one hour before the current time. This way, only the rows added in the last hour are uploaded to I2E.

This example shows how to use the select_data_added_between argument to only select the rows created within the DAG’s data interval (based on the created_at column in PostgreSQL). Note that the PostgreSQL table source_data_table must have a column named created_at for this to work.

from iqvia_nlp_provider.transfers.sql import SQLTimestampToI2EOperator

i2e_source_data_uri = SQLTimestampToI2EOperator(
    sql_conn_id="postgres_default",
    i2e_conn_id="i2e_default",
    sql_table="source_data_table",
    sql_timestamp_column="created_at",
    select_data_added_between=("{{ data_interval_start }}", "{{ data_interval_end }}"),
)

LocalFilesystemToI2EOperator¶

The LocalFilesystemToI2EOperator allows a local file to be uploaded to I2E as source data. It returns the URI of the uploaded source data in I2E.

This example shows how to use the LocalFilesystemToI2EOperator to upload a local file named /tmp/source.txt to I2E:

from iqvia_nlp_provider.transfers.local import LocalFilesystemToI2EOperator

i2e_source_data_uri = LocalFilesystemToI2EOperator(
    i2e_conn_id="i2e_default",
    local_file_path="/tmp/source.txt",
)

Transferring query results¶

I2EToS3Operator¶

The I2EToS3Operator allows I2E results to be written to an S3 bucket.

This example shows how to use the I2EToS3Operator to write I2E query results to the S3 bucket sdoh-query:

from iqvia_nlp_provider.operators.i2e import I2ERunQueryOperator
from iqvia_nlp_provider.transfers.s3 import I2EToS3Operator

i2e_query_results_uri = I2ERunQueryOperator(
    i2e_conn_id="i2e_default",
    i2e_index_uri="/api;type=index/sdoh-index",
    i2e_query_template="data_factory/sdoh-2_1_0/All Social Determinants of Health",
)

I2EToS3Operator(
    i2e_conn_id="i2e_default",
    aws_conn_id="aws_conn_id",
    i2e_query_results_uri=i2e_query_results_uri.output,
    aws_bucket_name="sdoh-query",
    aws_prefix="/results/",
    aws_filename="output.xml",
    s3_overwrite=True,
)

I2EToSQLOperator¶

The I2EToSQLOperator allows I2E results to be written to an SQL database. The operator is compatible with all database types supported by the apache-airflow-providers-common-sql provider, as listed in the Supported Database Types documentation. The I2EToSQLOperator has been tested against PostgreSQL, MySQL, Microsoft SQL Server, Snowflake, and Databricks.

To use the I2EToSQLOperator, I2E query results must be in a delimiter-separated format (e.g. TSV, CSV, PSV) and the i2e_results_delimiter argument must be set to the delimiter used in the query results.

For example, you could override the query settings with the I2ERunQueryOperator to set the output format to CSV, and then use the I2EToSQLOperator with i2e_results_delimiter="," to write the I2E query results to a PostgreSQL table named query_results_table:

from iqvia_nlp_provider.operators.i2e import I2ERunQueryOperator
from iqvia_nlp_provider.transfers.sql import I2EToSQLOperator

i2e_query_results_uri = I2ERunQueryOperator(
    i2e_conn_id="i2e_default",
    i2e_index_uri="/api;type=index/sdoh-index",
    i2e_query_template="data_factory/sdoh-2_1_0/All Social Determinants of Health",
    i2e_query_settings_overrides={"queryProperties": {"outputSettings": {"outputMode": "csv"}}},
)

I2EToSQLOperator(
    i2e_conn_id="i2e_default",
    sql_conn_id="postgres_default",
    i2e_query_results_uri=i2e_query_results_uri.output,
    i2e_results_delimiter=",",
    sql_table="query_results_table",
)

By default, all columns in the I2E query results are mapped to columns in the SQL table, and the SQL column names are auto-generated from the column names in the I2E query results. The column names are lower-cased and spaces replaced with underscores. For example, a column named Patient Age in the I2E query results will map to a column named patient_age in the SQL table.

The schema_mapping argument can be used to customize this mapping:

from iqvia_nlp_provider.transfers.sql import I2EToSQLOperator

I2EToSQLOperator(
    ...,
    schema_mapping={
        "Doc": "docs",
        "Date": "date",
        "Age": "age",
        "#Hits": "number_hits"
    },
    use_mapped_columns_only=True
)

In this example, the Doc column in the I2E query results will be mapped to the docs column in the SQL table. Similarly, the Date column in I2E will be mapped to the date column in SQL, and so on. Setting the use_mapped_columns_only argument to True means that only the columns specified in the schema_mapping are written to SQL, and any other columns in the I2E query results are ignored. If use_mapped_columns_only is set to False, then all columns from the I2E query results are written to SQL, and any I2E query result columns not present in schema_mapping are auto-converted into SQL column names using the rules described above.

Note

If you’re using schema_mapping and use_mapped_columns_only=False, make sure that the auto-generated column names don’t conflict with the mapped column names. For example, if you have an unmapped column in your I2E query results named “Patient Age”, and you explicitly map another column to “patient_age”, then this will cause a conflict as both will be mapped to the same column name.

The sql_audit_columns argument can be used to append extra column values to every inserted row. This is useful for tagging rows with batch or run metadata so that downstream SQL logic can identify and manage retry-induced duplicates. Keys are the target SQL column names; values are written as-is for each row. The parameter is templateable, so values can reference Airflow context:

from iqvia_nlp_provider.transfers.sql import I2EToSQLOperator

I2EToSQLOperator(
    ...,
    schema_mapping={
        "TF_DOCUMENT_GID": "DOCUMENT_GID",
        "ENTITYTYPE": "FLAG",
    },
    use_mapped_columns_only=True,
    sql_audit_columns={
        "RUN_ID": "{{ ti.xcom_pull(task_ids='get_unique_id') }}",
        "CREATE_TIME": "{{ ts }}",
    },
)

The target SQL table must already contain the audit columns. Audit column names must not collide with mapped result column names.

The operator does not remove duplicate rows on retry. Each DAG run inserts new rows; audit columns let you identify which run produced each row and deduplicate downstream in any SQL datastore.

Deduplicating retry-induced duplicates

When a batch-processing DAG is retried, the same I2E result can be written again with a new RUN_ID and CREATE_TIME. The business key—the columns that identify a unique result—is defined by your workflow. For example, you might use DOCUMENT_GID, FLAG, SENTENCE, and PHRASE as the business key, while RUN_ID and CREATE_TIME are metadata that distinguish insert attempts:

Example output table after two DAG retries¶
DOCUMENT_GID	FLAG	SENTENCE	PHRASE	RUN_ID	CREATE_TIME
1	Age	AMP will be 100 years old on 01-01-1990.	100 years old	manual__2026-06-25T14:48:00+00:00_abc	2026-06-25T14:48:35+00:00
1	Age	AMP will be 100 years old on 01-01-1990.	100 years old	manual__2026-06-25T14:57:00+00:00_def	2026-06-25T14:57:11+00:00

Both rows share the same business key but were produced by different runs. Downstream SQL may then be used to keep one row per business key (typically the latest by CREATE_TIME).

To list business keys that appear more than once:

SELECT DOCUMENT_GID, FLAG, SENTENCE, PHRASE, COUNT(*) AS row_count
FROM my_schema.my_output_table
GROUP BY DOCUMENT_GID, FLAG, SENTENCE, PHRASE
HAVING COUNT(*) > 1;

To expose deduplicated results, you could create a view that keeps the latest row per business key. This example uses ROW_NUMBER in a subquery; adjust PARTITION BY and ORDER BY to match your business key and tie-break rules:

CREATE OR REPLACE VIEW my_schema.my_output_table_deduped AS
SELECT
    DOCUMENT_GID,
    FLAG,
    SENTENCE,
    PHRASE,
    RUN_ID,
    CREATE_TIME
FROM (
    SELECT
        DOCUMENT_GID,
        FLAG,
        SENTENCE,
        PHRASE,
        RUN_ID,
        CREATE_TIME,
        ROW_NUMBER() OVER (
            PARTITION BY DOCUMENT_GID, FLAG, SENTENCE, PHRASE
            ORDER BY CREATE_TIME DESC, RUN_ID DESC
        ) AS row_num
    FROM my_schema.my_output_table
) ranked
WHERE row_num = 1;

On Snowflake, the same logic can be written more concisely with QUALIFY:

CREATE OR REPLACE VIEW my_schema.my_output_table_deduped AS
SELECT
    DOCUMENT_GID,
    FLAG,
    SENTENCE,
    PHRASE,
    RUN_ID,
    CREATE_TIME
FROM my_schema.my_output_table
QUALIFY ROW_NUMBER() OVER (
    PARTITION BY DOCUMENT_GID, FLAG, SENTENCE, PHRASE
    ORDER BY CREATE_TIME DESC, RUN_ID DESC
) = 1;

I2EToLocalFilesystemOperator¶

The I2EToLocalFilesystemOperator allows I2E results to be written to a local file.

This example shows how to use the I2EToLocalFilesystemOperator to write I2E query results to a local folder named /tmp/results/:

from iqvia_nlp_provider.operators.i2e import I2ERunQueryOperator
from iqvia_nlp_provider.transfers.local import I2EToLocalFilesystemOperator

i2e_query_results_uri = I2ERunQueryOperator(
    i2e_conn_id="i2e_default",
    i2e_index_uri="/api;type=index/sdoh-index",
    i2e_query_template="data_factory/sdoh-2_1_0/All Social Determinants of Health",
)

I2EToLocalFilesystemOperator(
    i2e_conn_id="i2e_default",
    i2e_query_results_uri=i2e_query_results_uri.output,
    local_folder="/tmp/results/",
)

Transferring resource bundles¶

S3ToI2EResourceBundleOperator¶

The S3ToI2EResourceBundleOperator retrieves an I2E resource bundle from an S3 bucket and installs the bundle across multiple I2E servers.

This example shows how to use the S3ToI2EResourceBundleOperator to retrieve an I2E resource bundle named sdoh_resource_bundle.zip from the S3 bucket sdoh-query, and install that bundle on the two I2E servers i2e_default_1 and i2e_default_2:

from iqvia_nlp_provider.transfers.s3 import S3ToI2EResourceBundleOperator

S3ToI2EResourceBundleOperator(
    i2e_conn_ids=["i2e_default_1", "i2e_default_2"],
    aws_conn_id="aws_conn_id",
    aws_bucket_name="sdoh-query",
    aws_key="sdoh_resource_bundle.zip",
)

This example shows how to use the get_all_i2e_connection_ids function to install the bundle on all I2E servers known to Airflow:

from iqvia_nlp_provider.transfers.s3 import S3ToI2EResourceBundleOperator
from iqvia_nlp_provider.utils.i2e_connections import get_all_i2e_connection_ids

S3ToI2EResourceBundleOperator(
    i2e_conn_ids=get_all_i2e_connection_ids(),
    aws_conn_id="aws_conn_id",
    aws_bucket_name="sdoh-query",
    aws_key="sdoh_resource_bundle.zip",
)

LocalFilesystemToI2EResourceBundleOperator¶

The LocalFilesystemToI2EResourceBundleOperator retrieves an I2E resource bundle from a local file and installs the bundle across multiple I2E servers.

This example shows how to use the LocalFilesystemToI2EResourceBundleOperator to retrieve an I2E resource bundle from a local file named /tmp/sdoh_resource_bundle.zip, and install that bundle on the two I2E servers i2e_default_1 and i2e_default_2:

from iqvia_nlp_provider.transfers.local import LocalFilesystemToI2EResourceBundleOperator

LocalFilesystemToI2EResourceBundleOperator(
    i2e_conn_ids=["i2e_default_1", "i2e_default_2"],
    local_file_path="/tmp/sdoh_resource_bundle.zip"
)

This example shows how to use the get_all_i2e_connection_ids function to install the bundle on all I2E servers known to Airflow:

from iqvia_nlp_provider.transfers.local import LocalFilesystemToI2EResourceBundleOperator
from iqvia_nlp_provider.utils.i2e_connections import get_all_i2e_connection_ids

LocalFilesystemToI2EResourceBundleOperator(
    i2e_conn_ids=get_all_i2e_connection_ids(),
    local_file_path="/tmp/sdoh_resource_bundle.zip"
)

Hooks¶

I2EHook¶

The I2EHook gives you more fine-grained control and customization when interacting with I2E. All of our operators use the I2EHook behind the scenes. Most of the time the operators are sufficient and you won’t need to use the I2EHook directly, but occasionally you might - especially if you’re doing something a bit non-standard.

You can use the I2EHook to upload source data to I2E. This could be useful for performing some pre-processing before uploading the source data.

from iqvia_nlp_provider.hooks.i2e import I2EHook

i2e = I2EHook(i2e_conn_id="i2e_default")
i2e_source_data_uri = i2e.upload_source_data(
    filename="source.txt",
    content=b"This is the content of my source data."
)

You can use the I2EHook to read query results from I2E. This could be useful for performing some post-processing on the query results, or for writing the query results to some external location that doesn’t have a corresponding transfer operator.

from iqvia_nlp_provider.hooks.i2e import I2EHook

i2e = I2EHook(i2e_conn_id="i2e_default")
results = i2e.get_query_results(query_results_uri="/api;type=server_tmp/my_query_results")
print(b"These are my query results: " + results.read())