Async API¶

Use AsyncIneepy for concurrent operations and better performance in async contexts.

When to use async¶

Concurrent requests: Make multiple requests simultaneously
High throughput: Handle many requests without blocking
Integration with async frameworks: FastAPI, aiohttp, asyncio applications
Long-running operations: Query or upload while doing other work

Basic usage¶

Context managerManual lifecycle

async def context_manager_example():
    async with AsyncIneepy(token=token) as client:
        user, etag = await client.users.get('me')
        print(f'Hello, {user.email}')


asyncio.run(context_manager_example())

async def manual_lifecycle_example():
    client = AsyncIneepy(token=token)
    try:
        user, etag = await client.users.get('me')
    finally:
        await client.aclose()


asyncio.run(manual_lifecycle_example())

Concurrent operations¶

Fetch multiple resources simultaneously:

async def fetch_all():
    async with AsyncIneepy(token=token) as client:
        user_coro = client.users.get('me')
        namespaces_coro = client.namespaces.list(
            namespace_id='raw.async_jobs'
        )
        user, user_etag = await user_coro
        namespaces = await namespaces_coro

        print(f'User: {user.email}')
        print(f'Namespaces: {len(namespaces.items)}')


asyncio.run(fetch_all())

Using asyncio.gather:

async def fetch_multiple_tables():
    async with AsyncIneepy(token=token) as client:
        table_ids = [
            'raw.async_jobs.events',
            'raw.async_jobs.events_b',
        ]
        tasks = [
            client.tables.get_metadata(table_id)
            for table_id in table_ids
        ]
        results = await asyncio.gather(*tasks)
        for table, etag in results:
            print(f'{table.id}: {len(table.columns)} columns')


asyncio.run(fetch_multiple_tables())

Streaming data¶

Upload and query large datasets:

async def large_query():
    async with AsyncIneepy(token=token) as client:
        result = await client.data.query(
            table_id='raw.async_jobs.events',
            format='json',
            timeout=600,
            poll_interval=2,
        )
        print(f'Rows: {len(result.items)}')


asyncio.run(large_query())

Async background jobs¶

Every write operation (submit_upload, submit_update) returns a job ID after transferring data to S3 — the server then processes the data asynchronously. In async code you can submit multiple jobs and poll them concurrently with asyncio.gather, which is faster than waiting for each one in sequence.

Submit and poll a single job¶

    async def submit_and_poll():
        data = _json.dumps([
            {'user_id': 2, 'event_type': 'signup'},
            {'user_id': 3, 'event_type': 'purchase'},
        ]).encode()

        async with AsyncIneepy(token=token) as client:
            job_id = await client.data.submit_upload(
                table_id='raw.async_jobs.events',
                data=data,
                format='json',
            )
            print(f'Job submitted: {job_id}')

            while True:
                job = await client.data.get_job(job_id)
                if job.status == _DataJobStatus.COMPLETED:
                    print('Upload complete')
                    break
                if job.status == _DataJobStatus.FAILED:
                    raise RuntimeError(job.error)
                await asyncio.sleep(2)

    asyncio.run(submit_and_poll())

Concurrent job polling with `asyncio.gather`¶

Submit two update jobs, then poll both at the same time. The inner poll coroutine loops until its job reaches a terminal state; asyncio.gather runs both polls concurrently on the same event loop:

    async def concurrent_uploads():
        batch_a = _json.dumps([
            {'user_id': 10, 'event_type': 'login'},
        ]).encode()
        batch_b = _json.dumps([
            {'user_id': 20, 'event_type': 'purchase'},
        ]).encode()

        async with AsyncIneepy(token=token) as client:
            job_id_a = await client.data.submit_update(
                table_id='raw.async_jobs.events',
                data=batch_a,
                format='json',
                update_mode='append',
            )
            job_id_b = await client.data.submit_update(
                table_id='raw.async_jobs.events',
                data=batch_b,
                format='json',
                update_mode='append',
            )

            async def poll(jid: str):
                while True:
                    job = await client.data.get_job(jid)
                    if job.status == _DataJobStatus.COMPLETED:
                        return job
                    if job.status == _DataJobStatus.FAILED:
                        raise RuntimeError(job.error)
                    await asyncio.sleep(2)

            job_a, job_b = await asyncio.gather(
                poll(job_id_a), poll(job_id_b)
            )
            print(
                f'Both complete: {job_a.status}, {job_b.status}'
            )

    asyncio.run(concurrent_uploads())

Tip

Use asyncio.gather to wait on multiple job IDs rather than await-ing them one at a time. With N jobs this cuts the wait time from N × processing-time to max(processing-times).

See Background jobs for a deeper look at job lifecycle, DataJobResponse fields, and error handling.

Integration with FastAPI¶

# Example: using AsyncIneepy in a FastAPI application
#
# from fastapi import FastAPI
# from ineepy import AsyncIneepy
# import os
#
# app = FastAPI()
# client = AsyncIneepy(token=os.getenv('INEEPY_TOKEN'))
#
# @app.on_event('shutdown')
# async def shutdown_event():
#     await client.aclose()
#
# @app.get('/user')
# async def get_user():
#     user, etag = await client.users.get('me')
#     return user.model_dump()

Error handling in async¶

async def safe_update():
    try:
        async with AsyncIneepy(token=token) as client:
            user, etag = await client.users.get('me')
            updated, new_etag = await client.users.update(
                'me',
                etag=etag,
                username=user.username,
            )
    except UnauthorizedError:
        print('Authentication failed')
    except ConflictError:
        print('Concurrent modification detected')


asyncio.run(safe_update())

Performance tips¶

Use asyncio.gather() for concurrent requests and job polling
Keep the client alive and reuse it across requests
Use async with to ensure proper resource cleanup
Set appropriate timeout and poll_interval for long operations
Limit concurrency — firing hundreds of requests at once can overwhelm the server. Use asyncio.Semaphore to cap the number of in-flight requests:

sem = asyncio.Semaphore(10)  # at most 10 concurrent requests

async def bounded(coro):
    async with sem:
        return await coro

results = await asyncio.gather(
    *[bounded(client.tables.get_metadata(tid)) for tid in table_ids]
)