Error handling¶

ineepy provides a structured exception hierarchy for error handling.

Exception hierarchy¶

IneepyError (base exception)
├── HTTPError
│   ├── ValidationError (400/422)
│   ├── UnauthorizedError (401)
│   ├── ForbiddenError (403)
│   ├── NotFoundError (404)
│   ├── ConflictError (409)
│   └── PreconditionFailedError (412)
├── JobFailedError
└── TimeoutError

Common exceptions¶

Exception	HTTP Status	When	Recovery
`ValidationError`	400/422	Invalid input	Check parameters
`UnauthorizedError`	401	Invalid credentials	Re-authenticate
`ForbiddenError`	403	Insufficient scope	Use admin scope if needed
`NotFoundError`	404	Resource not found	Check ID/path
`ConflictError`	409	Resource conflict	Check existing state
`PreconditionFailedError`	412	ETag mismatch	Fetch fresh and retry
`JobFailedError`	—	Async job failed	Check job.error
`TimeoutError`	—	Job didn't complete in time	Increase timeout or resubmit

Basic error handling¶

try:
    bad_token = get_token(
        email='user@ineep.org.br',
        password='wrongpassword',
        save_token=False,
    )
except UnauthorizedError:
    print('Invalid credentials')
except IneepyError as e:
    print(f'Error: {e}')

Handling specific errors¶

        try:
            client.namespaces.delete(namespace_id='raw.errtest')
        except ConflictError:
            print('Namespace is not empty. Delete tables first.')

ETag mismatch (optimistic locking)¶

When updating a resource, you must provide the current ETag. If the resource was modified by another client, you'll get PreconditionFailedError:

        try:
            user, etag = client.users.get('me')

            # Simulate a stale ETag by using a wrong value
            stale_etag = '"00000000-0000-0000-0000-000000000000"'
            updated, _ = client.users.update(
                'me',
                etag=stale_etag,
                username='newusername',
            )
        except PreconditionFailedError:
            print('ETag mismatch: another client modified the resource')
            # Fetch fresh and retry
            user, fresh_etag = client.users.get('me')
            updated, _ = client.users.update(
                'me',
                etag=fresh_etag,
                username=user.username,
            )
            print('Retry successful')

Async job errors¶

Some operations submit async jobs that may fail:

        try:
            job = client.data.upload(
                table_id='raw.errtest.events',
                data=b'[{"id": 1}, {"id": 2}]',
                format='json',
            )
            print(f'Upload succeeded: {job.status}')
        except JobFailedError as e:
            print(f'Job {e.job_id} failed: {e.error}')
        except TimeoutError as e:
            print(
                f'Job {e.job_id} timed out, '
                f'increase the timeout parameter'
            )

Error details¶

All HTTP errors include status code and detail:

        try:
            # Request a user that doesn't exist
            user, etag = client.users.get(999999)
        except HTTPError as e:
            print(f'Status: {e.status_code}')
            print(f'Detail: {e.detail}')

Retry patterns¶

Simple retry with exponential backoff¶

        def retry_with_backoff(func, max_attempts=3):
            for attempt in range(max_attempts):
                try:
                    return func()
                except IneepyError as e:
                    if attempt == max_attempts - 1:
                        raise
                    wait_time = 2 ** attempt
                    print(
                        f'Attempt {attempt + 1} failed, '
                        f'retrying in {wait_time}s'
                    )
                    time.sleep(wait_time)

        def my_operation():
            with Ineepy(token=token) as c:
                return c.users.get('me')

        try:
            result = retry_with_backoff(my_operation)
        except IneepyError:
            print('All retries failed')

Retry with exponential backoff (async)¶

async def retry_with_backoff_async(func, max_attempts=3):
    for attempt in range(max_attempts):
        try:
            return await func()
        except IneepyError:
            if attempt == max_attempts - 1:
                raise
            wait_time = 2 ** attempt
            print(
                f'Attempt {attempt + 1} failed, '
                f'retrying in {wait_time}s'
            )
            await asyncio.sleep(wait_time)


async def my_async_operation():
    async with AsyncIneepy(token=token) as client:
        return await client.users.get('me')


async def main():
    try:
        result = await retry_with_backoff_async(my_async_operation)
    except IneepyError:
        print('All retries failed')


asyncio.run(main())