print

k8s_job_executor

@resource
@contextmanager
def db_connection():
    try:
        db_conn = get_db_connection()
        yield db_conn
    finally:
        # crash here

dbt_cli_resource

snowflake_resource

s3_resource

s3_pickle_io_manager

def ingest_data(export_dict, host, collection_name, api_key):
    '''Downloads data from s3 based on the configurations in export_dict. Then iterates
    through each row of the downloaded data, writing them as documents into the specified
    Typesense collection.'''

    s3, response = get_files(export_dict)
    client = setup_typesense_client(host, api_key)
    start = time.time()

    if 'Contents' in response.keys():
        for item in response['Contents']:
            print('downloading file: ' + item['Key'])

            s3.download_file(EXPORT_BUCKET, item['Key'], 'tmp/address.csv'). # overwrites address.csv with each downloaded file

            print('finished downloading file')

            with open('tmp/address.csv') as f:
                reader = csv.DictReader(f)
                records = []
                i = 0
                # iterate through CSV, create Documents, write them into Typesense in batches of 100,000
                for row in reader:
                    record = create_document(row)
                    i += 1
                    records.append(record)
                    if len(records) == 100000:
                        # Write this batch of 100,000
                        print("Sending", len(records))
                        client.collections[collection_name].documents.import_(records, {'action': 'upsert'})
                        records = []
                # Write any leftovers that didn't make it up to 100,000
                print("Sending", len(records))
                client.collections[collection_name].documents.import_(records, {'action': 'upsert'})

    print('done in: ' + str(time.time() - start))

@op(required_resource_keys={'input_values','s3'})
def typesense_address_handler(context, logged_export_timestamp):
    # there is a lot of variable declaration stuff, and then:

    # Download exported S3 data and load it into Typesense.
    ingest_data(address_export_dict, ts_host, ts_collection, ts_key)