async def main():
    async with InfluxDBClientAsync(url='http://localhost:8086', token='my-token', org='my-org') as client:
        write_api = client.write_api()
        '\n        Async write\n        '

        async def async_write(batch):
            '\n            Prepare async task\n            '
            (await write_api.write(bucket='my-bucket', record=batch))
            return batch
        '\n        Prepare batches from generator\n        '
        batches = rx.from_iterable(csv_to_generator('vix-daily.csv')).pipe(ops.buffer_with_count(500)).pipe(ops.map((lambda batch: rx.from_future(asyncio.ensure_future(async_write(batch))))), ops.merge_all())
        done = asyncio.Future()
        '\n        Write batches by subscribing to Rx generator\n        '
        batches.subscribe(on_next=(lambda batch: print(f'Written batch... {len(batch)}')), on_error=(lambda ex: print(f'Unexpected error: {ex}')), on_completed=(lambda : done.set_result(0)), scheduler=AsyncIOScheduler(asyncio.get_event_loop()))
        '\n        Wait to finish all writes\n        '
        (await done)
async def main():
    async with InfluxDBClientAsync(url='http://localhost:8086', token='my-token', org='my-org') as client:
        write_api = client.write_api()
        '\n        Async write\n        '

        async def async_write(batch):
            '\n            Prepare async task\n            '
            (await write_api.write(bucket='my-bucket', record=batch))
            return batch
        '\n        Prepare batches from generator\n        '
        batches = rx.from_iterable(csv_to_generator('vix-daily.csv')).pipe(ops.buffer_with_count(500)).pipe(ops.map((lambda batch: rx.from_future(asyncio.ensure_future(async_write(batch))))), ops.merge_all())
        done = asyncio.Future()
        '\n        Write batches by subscribing to Rx generator\n        '
        batches.subscribe(on_next=(lambda batch: print(f'Written batch... {len(batch)}')), on_error=(lambda ex: print(f'Unexpected error: {ex}')), on_completed=(lambda : done.set_result(0)), scheduler=AsyncIOScheduler(asyncio.get_event_loop()))
        '\n        Wait to finish all writes\n        '
        (await done)