API Reference

Complete reference for the RestClient class. The documentation targets intapp-rest-client 1.x; the package is available on PyPI (opens in a new tab).

Constructor

from intapp_rest_client import RestClient
 
client = RestClient(
    base_url: str,
    # Authentication (pick one)
    oauth2_config: Optional[OAuth2Config] = None,
    entra_config: Optional[EntraConfig] = None,
    okta_config: Optional[OktaConfig] = None,
    api_key: Optional[str] = None,
    api_key_header: str = 'X-API-Key',
    api_key_in_query: bool = False,
    api_key_query_param: str = 'api_key',
    basic_auth: Optional[tuple] = None,
    jwt_config: Optional[JWTConfig] = None,
    httpx_auth: Optional[Any] = None,
    # Common settings
    timeout: float = 120.0,
    headers: Optional[Dict[str, str]] = None,
    retry_config: Optional[RetryConfig] = None,
    log_level: LogLevel = LogLevel.INFO,
    pool_config: Optional[ConnectionPoolConfig] = None,
    ssl_config: Optional[SSLConfig] = None,
    # Request tracking
    enable_request_id: bool = True,
    request_id_header: str = 'X-Request-ID',
    # Hooks
    pre_request_hook: Optional[Callable] = None,
    post_response_hook: Optional[Callable] = None,
)

Parameters

Parameter	Type	Default	Description
`base_url`	str	(required)	Base URL for all requests
`oauth2_config`	OAuth2Config	None	OAuth2 configuration
`entra_config`	EntraConfig	None	Microsoft Entra ID config
`okta_config`	OktaConfig	None	Okta configuration
`api_key`	str	None	API key for authentication
`api_key_header`	str	'X-API-Key'	Header name for API key
`api_key_in_query`	bool	False	Put API key in query params
`api_key_query_param`	str	'api_key'	Query param name
`basic_auth`	tuple	None	(username, password) tuple
`jwt_config`	JWTConfig	None	JWT configuration
`httpx_auth`	Any	None	External httpx-auth provider
`timeout`	float	120.0	Request timeout (seconds)
`headers`	dict	None	Default headers
`retry_config`	RetryConfig	None	Retry configuration
`log_level`	LogLevel	INFO	Logging level
`pool_config`	ConnectionPoolConfig	None	Connection pool config
`ssl_config`	SSLConfig	None	SSL/TLS configuration (custom certs, CA bundle, verify)
`enable_request_id`	bool	True	Add request ID header
`request_id_header`	str	'X-Request-ID'	Request ID header name
`pre_request_hook`	Callable	None	Pre-request callback
`post_response_hook`	Callable	None	Post-response callback

Synchronous Methods

get()

def get(
    endpoint: str,
    params: Optional[Dict[str, Any]] = None,
    headers: Optional[Dict[str, str]] = None,
    **kwargs
) -> Any

Make a GET request and return JSON response.

post()

def post(
    endpoint: str,
    json: Optional[Any] = None,
    data: Optional[Any] = None,
    params: Optional[Dict[str, Any]] = None,
    headers: Optional[Dict[str, str]] = None,
    **kwargs
) -> Any

Make a POST request and return JSON response.

put()

def put(
    endpoint: str,
    json: Optional[Any] = None,
    data: Optional[Any] = None,
    params: Optional[Dict[str, Any]] = None,
    headers: Optional[Dict[str, str]] = None,
    **kwargs
) -> Any

Make a PUT request and return JSON response.

patch()

def patch(
    endpoint: str,
    json: Optional[Any] = None,
    data: Optional[Any] = None,
    params: Optional[Dict[str, Any]] = None,
    headers: Optional[Dict[str, str]] = None,
    **kwargs
) -> Any

Make a PATCH request and return JSON response.

delete()

def delete(
    endpoint: str,
    json: Optional[Any] = None,
    params: Optional[Dict[str, Any]] = None,
    headers: Optional[Dict[str, str]] = None,
    **kwargs
) -> Any

Make a DELETE request and return JSON response.

head()

def head(
    endpoint: str,
    params: Optional[Dict[str, Any]] = None,
    headers: Optional[Dict[str, str]] = None,
    **kwargs
) -> Dict[str, str]

Make a HEAD request. Returns response headers as a dict (no body). Use for checking resource existence, Content-Length, or ETag.

options()

def options(
    endpoint: str,
    headers: Optional[Dict[str, str]] = None,
    **kwargs
) -> Dict[str, Any]

Make an OPTIONS request. Returns a dict with allowed_methods (list) and headers. Use for CORS or API discovery.

get_raw()

def get_raw(
    endpoint: str,
    params: Optional[Dict[str, Any]] = None,
    headers: Optional[Dict[str, str]] = None,
    **kwargs
) -> httpx.Response

Make a GET request and return raw response object. Use for binary downloads.

post_raw()

def post_raw(
    endpoint: str,
    json: Optional[Any] = None,
    data: Optional[Any] = None,
    files: Optional[Dict] = None,
    params: Optional[Dict[str, Any]] = None,
    headers: Optional[Dict[str, str]] = None,
    **kwargs
) -> httpx.Response

Make a POST request and return raw response object. Use for file uploads.

request_raw()

def request_raw(
    method: str,
    endpoint: str,
    **kwargs
) -> httpx.Response

Make a request with any HTTP method and return raw response.

download_file()

def download_file(
    endpoint: str,
    destination: Union[str, Path, BinaryIO],
    params: Optional[Dict[str, Any]] = None,
    headers: Optional[Dict[str, str]] = None,
    chunk_size: int = 8192,
    progress_callback: Optional[Callable[[int, Optional[int]], None]] = None,
    **kwargs
) -> int

Download a file with streaming (memory-efficient). Writes to a file path (str or Path) or a binary file-like object. Returns the number of bytes written. For large files, use this instead of get_raw() to avoid loading the entire response into memory. See Streaming file download for details.

Pagination Methods

get_paginated()

def get_paginated(
    endpoint: str,
    page_size: int,
    data_key: str,
    params: Optional[Dict[str, Any]] = None,
    headers: Optional[Dict[str, str]] = None,
    skip_param: str = "skip",
    limit_param: str = "limit",
    max_concurrency: int = 1,
    **kwargs
) -> Iterator[list]

Stream paginated GET responses. Yields batches of items. When max_concurrency > 1, pages are fetched in parallel but yielded in order.

post_paginated()

def post_paginated(
    endpoint: str,
    page_size: int,
    data_key: str,
    json: Optional[Dict[str, Any]] = None,
    params: Optional[Dict[str, Any]] = None,
    headers: Optional[Dict[str, str]] = None,
    skip_key: str = "skip",
    limit_key: str = "limit",
    max_concurrency: int = 1,
    **kwargs
) -> Iterator[list]

Stream paginated POST responses (pagination in body). When max_concurrency > 1, pages are fetched in parallel but yielded in order.

post_paginated_params()

def post_paginated_params(
    endpoint: str,
    page_size: int,
    data_key: str,
    json: Optional[Dict[str, Any]] = None,
    params: Optional[Dict[str, Any]] = None,
    headers: Optional[Dict[str, str]] = None,
    skip_param: str = "skip",
    limit_param: str = "limit",
    max_concurrency: int = 1,
    **kwargs
) -> Iterator[list]

Stream paginated POST responses (pagination in query params). When max_concurrency > 1, pages are fetched in parallel but yielded in order.

Parallel Methods

Execute multiple requests in parallel using a thread pool. Use for bulk GET/POST/PUT/DELETE operations. See Parallel requests for examples.

parallel_get()

def parallel_get(
    endpoints: List[str],
    max_workers: int = 8,
    error_handling: ErrorHandling = ErrorHandling.FAIL_FAST,
    **kwargs
) -> Union[List[Any], Tuple[List[Any], List[Dict]]]

Execute multiple GET requests in parallel. endpoints: list of paths or full URLs. error_handling: FAIL_FAST (raise on first error), COLLECT (return results and errors), or LOG_ONLY (log and continue). With FAIL_FAST, returns a list of JSON responses. With COLLECT or LOG_ONLY, returns (results, errors).

parallel_post()

def parallel_post(
    endpoint: str,
    payloads: List[Dict[str, Any]],
    max_workers: int = 8,
    error_handling: ErrorHandling = ErrorHandling.FAIL_FAST,
    **kwargs
) -> Union[List[Any], Tuple[List[Any], List[Dict]]]

Execute multiple POST requests to the same endpoint in parallel. payloads: one JSON body per request. Return shape same as parallel_get.

parallel_put()

def parallel_put(
    endpoint: str,
    payloads: List[Dict[str, Any]],
    max_workers: int = 8,
    error_handling: ErrorHandling = ErrorHandling.FAIL_FAST,
    **kwargs
) -> Union[List[Any], Tuple[List[Any], List[Dict]]]

Execute multiple PUT requests to the same endpoint in parallel. payloads: one JSON body per request. Return shape same as parallel_get.

parallel_delete()

def parallel_delete(
    endpoint: str,
    payloads: List[Dict[str, Any]],
    max_workers: int = 8,
    error_handling: ErrorHandling = ErrorHandling.FAIL_FAST,
    **kwargs
) -> Union[List[Any], Tuple[List[Any], List[Dict]]]

Execute multiple DELETE requests to the same endpoint in parallel, with one JSON body per request (e.g. batch delete). Return shape same as parallel_get.

Asynchronous Methods

All synchronous methods have async equivalents prefixed with a:

Sync	Async
`get()`	`aget()`
`post()`	`apost()`
`put()`	`aput()`
`patch()`	`apatch()`
`delete()`	`adelete()`
`head()`	`ahead()`
`options()`	`aoptions()`
`get_raw()`	`aget_raw()`
`post_raw()`	`apost_raw()`
`request_raw()`	`arequest_raw()`
`download_file()`	`adownload_file()`
`get_paginated()`	`aget_paginated()`
`post_paginated()`	`apost_paginated()`
`post_paginated_params()`	`apost_paginated_params()`
`parallel_get()`	`aparallel_get()`
`parallel_post()`	`aparallel_post()`
`parallel_put()`	`aparallel_put()`
`parallel_delete()`	`aparallel_delete()`
`close()`	`aclose()`

Async parallel methods

Async parallel helpers are async iterators that yield (index, result, error) tuples as work completes (see docstrings in code for full behavior). They differ from the sync parallel_* methods, which return lists (or (results, errors) tuples).

async def aparallel_get(
    endpoints: List[str],
    max_concurrency: int = 8,
    error_handling: ErrorHandling = ErrorHandling.FAIL_FAST,
    **kwargs
) -> AsyncIterator[Tuple[int, Optional[Dict[str, Any]], Optional[Exception]]]

aparallel_post, aparallel_put, and aparallel_delete use the same yield shape (index, JSON result or None, exception or None). Prefer Parallel requests for synchronous batch APIs; use aparallel_* for async ordered iteration over parallel results.

Utility Methods

generate_request_id()

def generate_request_id() -> str

Generate a unique UUID for distributed tracing.

build_url()

def build_url(endpoint: str) -> str

Build a full URL from an endpoint path.

get_auth_headers()

def get_auth_headers() -> Dict[str, str]

Get current auth headers synchronously.

aget_auth_headers()

async def aget_auth_headers() -> Dict[str, str]

Get current auth headers asynchronously.

sanitize_headers()

def sanitize_headers(headers: Dict[str, str]) -> Dict[str, str]

Mask sensitive header values for safe logging.

Lifecycle Methods

close()

def close() -> None

Close the synchronous HTTP client and release connections.

aclose()

async def aclose() -> None

Close the asynchronous HTTP client and release connections.

Context Managers

Synchronous

with RestClient(base_url="https://api.example.com", api_key="key") as client:
    data = client.get("/api/resources")
# Client automatically closed

Asynchronous

async with RestClient(base_url="https://api.example.com", api_key="key") as client:
    data = await client.aget("/api/resources")
# Client automatically closed

Usage Examples

Basic Request

client = RestClient(base_url="https://api.example.com", api_key="key")
data = client.get("/api/users", params={"limit": 100})
client.close()

With Context Manager

with RestClient(base_url="https://api.example.com", api_key="key") as client:
    users = client.get("/api/users")
    companies = client.get("/api/companies")

Async with Pagination

async with RestClient(base_url="https://api.example.com", api_key="key") as client:
    async for batch in client.aget_paginated("/api/items", 1000, "items"):
        await process_batch(batch)

File Upload

with open("document.pdf", "rb") as f:
    response = client.post_raw(
        "/api/upload",
        files={"file": ("document.pdf", f, "application/pdf")}
    )

File Download (raw)

response = client.get_raw("/api/files/123")
with open("output.pdf", "wb") as f:
    f.write(response.content)

Streaming File Download

# Memory-efficient: streams to file without loading into memory
size = client.download_file("/api/files/123", "/tmp/output.pdf")
# Or to a file object
with open("/tmp/output.pdf", "wb") as f:
    client.download_file("/api/files/123", f)

Exceptions Changelog