Async API¶

Every sync service has an Async* counterpart with the same method surface, importable from congressgov.async_api:

from congressgov.async_api import AsyncBill

bill = await AsyncBill(client=async_client).get(congress=118, bill_type="hr", bill_number=1)

Async Congress.gov services.

from congressgov.async_api import AsyncBill, AsyncMember

AsyncAmendment ¶

AsyncAmendment(client: 'AuthenticatedClient | None' = None)

Bases: AsyncApiService

Amendment service provides API access for fetching Congressional amendment data.

This service class handles API interactions for fetching amendments from the Congress.gov API. Instance methods on the Amendment model (registered in congressgov.services.extensions.amendments) provide additional functionality.

Separation of Concerns: - This service class: Fetches data from API (get, search) - Amendment model extensions: Operate on fetched data (expand, get_actions, etc.) - Amendments model extensions: Query and filter collections (filter, by_type, etc.)

USAGE EXAMPLES

Fetch an amendment¶

amendment_service = Amendment(client=my_client) amendment = amendment_service.get(congress=118, amendment_type="hamdt", amendment_number="1")

Use extension methods¶

expanded = amendment.expand(attributes=['actions', 'cosponsors']) actions = amendment.get_actions()

Search for amendments¶

amendments = amendment_service.search(congress=118, limit=10) house_amdts = amendments.house_amendments()

Initialize Amendment service.

Source code in src/congressgov/services/async_api/amendment.py

def __init__(self, client: 'AuthenticatedClient | None' = None) -> None:
    """Initialize Amendment service."""
    self.client = client
    self._last_result: 'AmendmentModel | None' = None

get `async` ¶

get(*, client: 'AuthenticatedClient | None' = None, congress: int, amendment_type: str, amendment_number: str, format_: str | None = None) -> 'AmendmentModel'

Retrieve a single amendment's details from the API.

Parameters:

Name	Type	Description	Default
`client`	`'AuthenticatedClient \| None'`	The API client instance.	`None`
`congress`	`int`	Congress number.	required
`amendment_type`	`str`	Amendment type (e.g., 'hamdt', 'samdt').	required
`amendment_number`	`str`	Amendment number.	required
`format_`	`str \| None`	Response format.	`None`

Returns:

Name	Type	Description
`Amendment`	`'AmendmentModel'`	Parsed Amendment model with client attached.

Raises:

Type	Description
`ValidationError`	If parameters are invalid (includes suggestions)
`APIError`	If API request fails
`ClientNotFoundError`	If no client is available

Example

amendment_service = Amendment(client=my_client) amendment = amendment_service.get(congress=118, amendment_type="hamdt", amendment_number="1") expanded = amendment.expand()

Source code in src/congressgov/services/async_api/amendment.py

async def get(
    self,
    *,
    client: 'AuthenticatedClient | None' = None,
    congress: int,
    amendment_type: str,
    amendment_number: str,
    format_: str | None = None
) -> 'AmendmentModel':
    """
    Retrieve a single amendment's details from the API.

    Args:
        client: The API client instance.
        congress: Congress number.
        amendment_type: Amendment type (e.g., 'hamdt', 'samdt').
        amendment_number: Amendment number.
        format_: Response format.

    Returns:
        Amendment: Parsed Amendment model with client attached.

    Raises:
        ValidationError: If parameters are invalid (includes suggestions)
        APIError: If API request fails
        ClientNotFoundError: If no client is available

    Example:
        >>> amendment_service = Amendment(client=my_client)
        >>> amendment = amendment_service.get(congress=118, amendment_type="hamdt", amendment_number="1")
        >>> expanded = amendment.expand()
    """
    # --- <VALIDATE PARAMETERS> ---
    validate_congress(congress)
    validate_amendment_type(amendment_type)

    resolved_client = await AsyncApiService._resolve_client(self, client)

    resp = await amendment_details_async(
        client=resolved_client,
        congress=congress,
        amendment_type=amendment_type,
        amendment_number=amendment_number,
        format_=format_,
    )

    api_env = ApiEnvelope.model_validate(json.loads(resp.content))
    result = AmendmentModel.model_validate(api_env.data)

    # Attach client for extension methods
    result.client = resolved_client
    self._last_result = result

    return result

search `async` ¶

search(*, client: 'AuthenticatedClient | None' = None, congress: int | None = None, amendment_type: str | None = None, format_: str | None = None, offset: int | None = None, limit: int | None = None, from_date_time: str | None = None, to_date_time: str | None = None, sort: str | None = None) -> 'AmendmentsModel'

Search for amendments using the universal search system.

Parameters:

Name	Type	Description	Default
`client`	`'AuthenticatedClient \| None'`	The API client instance.	`None`
`congress`	`int \| None`	Congress number (optional).	`None`
`amendment_type`	`str \| None`	Amendment type (optional).	`None`
`format_`	`str \| None`	Response format.	`None`
`offset`	`int \| None`	Offset for pagination.	`None`
`limit`	`int \| None`	Limit for pagination.	`None`
`from_date_time`	`str \| None`	Start date/time filter.	`None`
`to_date_time`	`str \| None`	End date/time filter.	`None`
`sort`	`str \| None`	Sort order.	`None`

Returns:

Name	Type	Description
`Amendments`	`'AmendmentsModel'`	Parsed Amendments collection model.

Example

amendment_service = Amendment(client=my_client) amendments = amendment_service.search(congress=118, limit=50) house_amdts = amendments.house_amendments()

Source code in src/congressgov/services/async_api/amendment.py

async def search(
    self,
    *,
    client: 'AuthenticatedClient | None' = None,
    congress: int | None = None,
    amendment_type: str | None = None,
    format_: str | None = None,
    offset: int | None = None,
    limit: int | None = None,
    from_date_time: str | None = None,
    to_date_time: str | None = None,
    sort: str | None = None
) -> 'AmendmentsModel':
    """
    Search for amendments using the universal search system.

    Args:
        client: The API client instance.
        congress: Congress number (optional).
        amendment_type: Amendment type (optional).
        format_: Response format.
        offset: Offset for pagination.
        limit: Limit for pagination.
        from_date_time: Start date/time filter.
        to_date_time: End date/time filter.
        sort: Sort order.

    Returns:
        Amendments: Parsed Amendments collection model.

    Example:
        >>> amendment_service = Amendment(client=my_client)
        >>> amendments = amendment_service.search(congress=118, limit=50)
        >>> house_amdts = amendments.house_amendments()
    """
    from congressgov.services.core.search import search_async as universal_search

    return await universal_search(
        'amendment',
        self,
        client=client,
        congress=congress,
        amendment_type=amendment_type,
        format_=format_,
        offset=offset,
        limit=limit,
        from_date_time=from_date_time,
        to_date_time=to_date_time,
        sort=sort
    )

AsyncBill ¶

AsyncBill(client: 'AuthenticatedClient | None' = None)

Bases: AsyncApiService

Async Bill service provides non-blocking API access for fetching Congressional bill data.

This async service class mirrors the sync Bill service but uses async/await for non-blocking operations. It handles API interactions asynchronously for fetching bills from the Congress.gov API.

Separation of Concerns: - This service class: Fetches data from API asynchronously (get, search) - Bill model extensions: Operate on fetched data (expand_async, get_actions_async, etc.) - Bills model extensions: Query and filter collections (filter, by_type, etc.)

Key Benefits: - Non-blocking API calls allow concurrent operations - Efficient batch processing with asyncio.gather() - Streaming support for large datasets - Full compatibility with async frameworks (FastAPI, Django async)

USAGE EXAMPLES

Basic async usage - fetch a bill¶

async_bill_service = AsyncBill(client=my_client) bill = await async_bill_service.get(congress=118, bill_type="hr", bill_number=1)

Fetch multiple bills concurrently¶

tasks = [ async_bill_service.get(congress=118, bill_type="hr", bill_number=i) for i in range(1, 11) ] bills = await asyncio.gather(*tasks)

Search for bills¶

bills = await async_bill_service.search(congress=118, bill_type="hr", limit=10)

Stream large datasets¶

async for batch in async_bill_service.search_stream(congress=118, batch_size=50): for bill in batch.bills: # Process each bill without loading all into memory await process_bill(bill)

Initialize AsyncBill service.

Parameters:

Name	Type	Description	Default
`client`	`'AuthenticatedClient \| None'`	Optional API client instance. Can also be provided per-method.	`None`

Source code in src/congressgov/services/async_api/bill.py

def __init__(self, client: 'AuthenticatedClient | None' = None) -> None:
    """
    Initialize AsyncBill service.

    Args:
        client: Optional API client instance. Can also be provided per-method.
    """
    self.client = client
    self._last_result: 'BillModel | None' = None  # Store last fetched bill

get `async` ¶

get(*, client: 'AuthenticatedClient | None' = None, congress: int, bill_type: str, bill_number: int, format_: str | None = None) -> 'BillModel'

Retrieve a single bill's details from the API asynchronously.

Parameters:

Name	Type	Description	Default
`client`	`'AuthenticatedClient \| None'`	The API client instance. If None, will use self.client if available.	`None`
`congress`	`int`	Congress number (e.g., 118 for 118th Congress).	required
`bill_type`	`str`	Bill type (e.g., 'hr', 's', 'hjres', 'sjres').	required
`bill_number`	`int`	Bill number.	required
`format_`	`str \| None`	Response format (default: json).	`None`

Returns:

Name	Type	Description
`Bill`	`'BillModel'`	Parsed Bill model instance from API response with client attached.

Raises:

Type	Description
`ValidationError`	If parameters are invalid (includes suggestions)
`APIError`	If API request fails
`ClientNotFoundError`	If no client is available

Example

async_bill_service = AsyncBill(client=my_client) bill = await async_bill_service.get(congress=118, bill_type="hr", bill_number=1)

Now use async extension methods¶

expanded_bill = await bill.expand_async() actions = await bill.get_actions_async()

Source code in src/congressgov/services/async_api/bill.py

async def get(
    self,
    *,
    client: 'AuthenticatedClient | None' = None,
    congress: int,
    bill_type: str,
    bill_number: int,
    format_: str | None = None
) -> 'BillModel':
    """
    Retrieve a single bill's details from the API asynchronously.

    Args:
        client: The API client instance. If None, will use self.client if available.
        congress: Congress number (e.g., 118 for 118th Congress).
        bill_type: Bill type (e.g., 'hr', 's', 'hjres', 'sjres').
        bill_number: Bill number.
        format_: Response format (default: json).

    Returns:
        Bill: Parsed Bill model instance from API response with client attached.

    Raises:
        ValidationError: If parameters are invalid (includes suggestions)
        APIError: If API request fails
        ClientNotFoundError: If no client is available

    Example:
        >>> async_bill_service = AsyncBill(client=my_client)
        >>> bill = await async_bill_service.get(congress=118, bill_type="hr", bill_number=1)
        >>> # Now use async extension methods
        >>> expanded_bill = await bill.expand_async()
        >>> actions = await bill.get_actions_async()
    """
    # --- <VALIDATE PARAMETERS> ---
    validate_congress(congress)
    validate_bill_type(bill_type)

    # --- <RESOLVE CLIENT PARAMETER> ---
    resolved_client = await AsyncApiService._resolve_client(self, client)

    # --- <CALL ASYNC API> ---
    resp = await bill_details_async(
        congress=congress,
        bill_type=bill_type,
        bill_number=bill_number,
        client=resolved_client,
        format_=format_ or 'json'
    )

    # --- <PARSE AND RETURN> ---
    api_env = ApiEnvelope.model_validate(json.loads(resp.content))
    bill = BillModel.model_validate(api_env.data)

    # Attach client for extension methods
    bill.client = resolved_client

    # Store last result
    self._last_result = bill

    return bill

search `async` ¶

search(*, client: 'AuthenticatedClient | None' = None, congress: int | None = None, bill_type: str | None = None, format_: str | None = None, offset: int | None = None, limit: int | None = None, from_date_time: str | None = None, to_date_time: str | None = None, sort: str | None = None) -> 'BillsModel'

Search for bills using the universal search system asynchronously.

This method provides flexible searching across bills with optional filters.

Parameters:

Name	Type	Description	Default
`client`	`'AuthenticatedClient \| None'`	The API client instance. If None, will use self.client if available.	`None`
`congress`	`int \| None`	Congress number to filter by (e.g., 118 for 118th Congress).	`None`
`bill_type`	`str \| None`	Bill type to filter by. Valid values: - 'hr': House Bill - 's': Senate Bill - 'hjres': House Joint Resolution - 'sjres': Senate Joint Resolution - 'hconres': House Concurrent Resolution - 'sconres': Senate Concurrent Resolution - 'hres': House Simple Resolution - 'sres': Senate Simple Resolution	`None`
`format_`	`str \| None`	Response format (default: json).	`None`
`offset`	`int \| None`	Pagination offset (number of records to skip).	`None`
`limit`	`int \| None`	Maximum number of results to return (default: 20).	`None`
`from_date_time`	`str \| None`	Start date/time for filtering (ISO 8601 format).	`None`
`to_date_time`	`str \| None`	End date/time for filtering (ISO 8601 format).	`None`
`sort`	`str \| None`	Sort order (e.g., 'updateDate+desc').	`None`

Returns:

Name	Type	Description
`Bills`	`'BillsModel'`	Collection of bills matching the search criteria with attached client
	`'BillsModel'`	for extension methods.

Raises:

Type	Description
`ValidationError`	If parameters are invalid (includes suggestions)
`APIError`	If API request fails
`ClientNotFoundError`	If no client is available

Example

async_bill_service = AsyncBill(client=my_client)

Search for House bills from 118th Congress¶

bills = await async_bill_service.search(congress=118, bill_type="hr", limit=50) print(f"Found {len(bills.bills)} House bills")

Search with date filters¶

bills = await async_bill_service.search( ... congress=118, ... from_date_time="2023-01-01T00:00:00Z", ... to_date_time="2023-12-31T23:59:59Z", ... limit=100 ... )

Source code in src/congressgov/services/async_api/bill.py

async def search(
    self,
    *,
    client: 'AuthenticatedClient | None' = None,
    congress: int | None = None,
    bill_type: str | None = None,
    format_: str | None = None,
    offset: int | None = None,
    limit: int | None = None,
    from_date_time: str | None = None,
    to_date_time: str | None = None,
    sort: str | None = None
) -> 'BillsModel':
    """
    Search for bills using the universal search system asynchronously.

    This method provides flexible searching across bills with optional filters.

    Args:
        client: The API client instance. If None, will use self.client if available.
        congress: Congress number to filter by (e.g., 118 for 118th Congress).
        bill_type: Bill type to filter by. Valid values:
            - 'hr': House Bill
            - 's': Senate Bill
            - 'hjres': House Joint Resolution
            - 'sjres': Senate Joint Resolution
            - 'hconres': House Concurrent Resolution
            - 'sconres': Senate Concurrent Resolution
            - 'hres': House Simple Resolution
            - 'sres': Senate Simple Resolution
        format_: Response format (default: json).
        offset: Pagination offset (number of records to skip).
        limit: Maximum number of results to return (default: 20).
        from_date_time: Start date/time for filtering (ISO 8601 format).
        to_date_time: End date/time for filtering (ISO 8601 format).
        sort: Sort order (e.g., 'updateDate+desc').

    Returns:
        Bills: Collection of bills matching the search criteria with attached client
        for extension methods.

    Raises:
        ValidationError: If parameters are invalid (includes suggestions)
        APIError: If API request fails
        ClientNotFoundError: If no client is available

    Example:
        >>> async_bill_service = AsyncBill(client=my_client)
        >>> 
        >>> # Search for House bills from 118th Congress
        >>> bills = await async_bill_service.search(congress=118, bill_type="hr", limit=50)
        >>> print(f"Found {len(bills.bills)} House bills")
        >>> 
        >>> # Search with date filters
        >>> bills = await async_bill_service.search(
        ...     congress=118,
        ...     from_date_time="2023-01-01T00:00:00Z",
        ...     to_date_time="2023-12-31T23:59:59Z",
        ...     limit=100
        ... )
    """
    # --- <VALIDATE PARAMETERS> ---
    if congress is not None:
        validate_congress(congress)
    if bill_type is not None:
        validate_bill_type(bill_type)

    # --- <RESOLVE CLIENT PARAMETER> ---
    resolved_client = await AsyncApiService._resolve_client(self, client)

    # --- <DETERMINE WHICH SEARCH FUNCTION TO USE> ---
    if congress and bill_type:
        # Search by both congress and type
        resp = await bill_list_by_type_async(
            congress=congress,
            bill_type=bill_type,
            client=resolved_client,
            format_=format_ or 'json',
            offset=offset,
            limit=limit,
            from_date_time=from_date_time,
            to_date_time=to_date_time,
            sort=sort
        )
    elif congress:
        # Search by congress only
        resp = await bill_list_by_congress_async(
            congress=congress,
            client=resolved_client,
            format_=format_ or 'json',
            offset=offset,
            limit=limit,
            from_date_time=from_date_time,
            to_date_time=to_date_time,
            sort=sort
        )
    else:
        # Search all bills
        resp = await bill_list_all_async(
            client=resolved_client,
            format_=format_ or 'json',
            offset=offset,
            limit=limit,
            from_date_time=from_date_time,
            to_date_time=to_date_time,
            sort=sort
        )

    # --- <PARSE AND RETURN> ---
    api_env = ApiEnvelope.model_validate(json.loads(resp.content))
    bills = BillsModel.model_validate(api_env.data)

    # Attach client for extension methods
    bills.client = resolved_client
    propagate_client_to_items(bills, "bills", resolved_client)

    return bills

search_stream `async` ¶

search_stream(*, congress: int | None = None, bill_type: str | None = None, batch_size: int = 100, **kwargs: Any) -> AsyncIterator['BillsModel']

Stream bills in batches for memory-efficient processing of large datasets.

This method yields batches of bills without loading the entire result set into memory at once. Ideal for processing thousands of bills.

Parameters:

Name	Type	Description	Default
`congress`	`int \| None`	Congress number to filter by.	`None`
`bill_type`	`str \| None`	Bill type to filter by.	`None`
`batch_size`	`int`	Number of bills per batch (default: 100).	`100`
`**kwargs`	`Any`	Additional search parameters (from_date_time, to_date_time, etc.)	`{}`

Yields:

Name	Type	Description
`BillsModel`	`AsyncIterator['BillsModel']`	Batches of bills

Example

async_bill_service = AsyncBill(client=my_client)

Stream and process bills in batches¶

async for batch in async_bill_service.search_stream( ... congress=118, ... batch_size=50 ... ): ... for bill in batch.bills: ... # Process each bill ... await process_bill(bill) ... print(f"Processed batch of {len(batch.bills)} bills")

Source code in src/congressgov/services/async_api/bill.py

async def search_stream(
    self,
    *,
    congress: int | None = None,
    bill_type: str | None = None,
    batch_size: int = 100,
    **kwargs: Any
) -> AsyncIterator['BillsModel']:
    """
    Stream bills in batches for memory-efficient processing of large datasets.

    This method yields batches of bills without loading the entire result set
    into memory at once. Ideal for processing thousands of bills.

    Args:
        congress: Congress number to filter by.
        bill_type: Bill type to filter by.
        batch_size: Number of bills per batch (default: 100).
        **kwargs: Additional search parameters (from_date_time, to_date_time, etc.)

    Yields:
        BillsModel: Batches of bills

    Example:
        >>> async_bill_service = AsyncBill(client=my_client)
        >>> 
        >>> # Stream and process bills in batches
        >>> async for batch in async_bill_service.search_stream(
        ...     congress=118,
        ...     batch_size=50
        ... ):
        ...     for bill in batch.bills:
        ...         # Process each bill
        ...         await process_bill(bill)
        ...     print(f"Processed batch of {len(batch.bills)} bills")
    """
    offset = 0
    stream_kwargs = dict(kwargs)
    stream_kwargs.pop("limit", None)
    while True:
        batch = await self.search(
            congress=congress,
            bill_type=bill_type,
            offset=offset,
            limit=batch_size,
            **stream_kwargs,
        )

        if not batch.bills or len(batch.bills) == 0:
            break

        yield batch
        if len(batch.bills) < batch_size:
            break
        offset += batch_size

search_by_laws `async` ¶

search_by_laws(*, client: 'AuthenticatedClient | None' = None, congress: int | None = None, law_type: str | None = None, law_number: int | None = None, format_: str | None = None, offset: int | None = None, limit: int | None = None) -> 'BillModel | BillsModel'

Search for bills by public or private law asynchronously.

Parameters:

Name	Type	Description	Default
`client`	`'AuthenticatedClient \| None'`	The API client instance.	`None`
`congress`	`int \| None`	Congress number.	`None`
`law_type`	`str \| None`	Law type ('pub' or 'priv').	`None`
`law_number`	`int \| None`	Specific law number.	`None`
`format_`	`str \| None`	Response format.	`None`
`offset`	`int \| None`	Pagination offset.	`None`
`limit`	`int \| None`	Maximum results.	`None`

Returns:

Type	Description
`'BillModel \| BillsModel'`	BillModel or BillsModel depending on search specificity.

Example

Get specific public law¶

bill = await async_bill_service.search_by_laws( ... congress=117, ... law_type="pub", ... law_number=108 ... )

Source code in src/congressgov/services/async_api/bill.py

async def search_by_laws(
    self,
    *,
    client: 'AuthenticatedClient | None' = None,
    congress: int | None = None,
    law_type: str | None = None,
    law_number: int | None = None,
    format_: str | None = None,
    offset: int | None = None,
    limit: int | None = None
) -> 'BillModel | BillsModel':
    """
    Search for bills by public or private law asynchronously.

    Args:
        client: The API client instance.
        congress: Congress number.
        law_type: Law type ('pub' or 'priv').
        law_number: Specific law number.
        format_: Response format.
        offset: Pagination offset.
        limit: Maximum results.

    Returns:
        BillModel or BillsModel depending on search specificity.

    Example:
        >>> # Get specific public law
        >>> bill = await async_bill_service.search_by_laws(
        ...     congress=117,
        ...     law_type="pub",
        ...     law_number=108
        ... )
    """
    # --- <VALIDATE PARAMETERS> ---
    if congress is not None:
        validate_congress(congress)

    # --- <RESOLVE CLIENT> ---
    resolved_client = await AsyncApiService._resolve_client(self, client)

    # --- <CALL ASYNC API> ---
    if congress and law_type and law_number:
        # Specific law
        resp = await law_list_by_congress_law_type_and_law_number_async(
            congress=congress,
            law_type=law_type,
            law_number=law_number,
            client=resolved_client,
            format_=format_ or 'json'
        )
        api_env = ApiEnvelope.model_validate(json.loads(resp.content))
        bill = BillModel.model_validate(api_env.data)
        bill.client = resolved_client
        return bill
    elif congress and law_type:
        # By congress and law type
        resp = await law_list_by_congress_and_law_type_async(
            congress=congress,
            law_type=law_type,
            client=resolved_client,
            format_=format_ or 'json',
            offset=offset,
            limit=limit
        )
    elif congress:
        # By congress only
        resp = await law_list_by_congress_async(
            congress=congress,
            client=resolved_client,
            format_=format_ or 'json',
            offset=offset,
            limit=limit
        )
    else:
        raise ValueError("At least 'congress' parameter must be provided")

    # --- <PARSE AND RETURN> ---
    api_env = ApiEnvelope.model_validate(json.loads(resp.content))
    bills = BillsModel.model_validate(api_env.data)
    bills.client = resolved_client
    propagate_client_to_items(bills, "bills", resolved_client)
    return bills

AsyncBoundCongressionalRecord ¶

AsyncBoundCongressionalRecord(client: AuthenticatedClient | None = None)

Bases: AsyncApiService

BoundCongressionalRecord service provides API access for bound congressional record data.

Source code in src/congressgov/services/async_api/bound_congressional_record.py

def __init__(self, client: AuthenticatedClient | None = None):
    self.client = client

search `async` ¶

search(*, client: AuthenticatedClient | None = None, year: int = None, month: int = None, format_: str = None, offset: int = None, limit: int = None)

Search for bound congressional records using the universal search system.

Source code in src/congressgov/services/async_api/bound_congressional_record.py

async def search(
    self,
    *,
    client: AuthenticatedClient | None = None,
    year: int = None,
    month: int = None,
    format_: str = None,
    offset: int = None,
    limit: int = None
):
    """Search for bound congressional records using the universal search system."""
    from congressgov.services.core.search import search_async as universal_search

    return await universal_search(
        'bound-congressional-record',
        self,
        client=client,
        year=year,
        month=month,
        format_=format_,
        offset=offset,
        limit=limit
    )

AsyncCRSReport ¶

AsyncCRSReport(client: AuthenticatedClient | None = None)

Bases: AsyncApiService

CRSReport service provides API access for CRS report data.

Source code in src/congressgov/services/async_api/crsreport.py

def __init__(self, client: AuthenticatedClient | None = None):
    self.client = client
    self._last_result = None

search `async` ¶

search(*, client: AuthenticatedClient | None = None, format_: str = None, offset: int = None, limit: int = None)

Search for CRS reports using the universal search system.

Source code in src/congressgov/services/async_api/crsreport.py

async def search(
    self,
    *,
    client: AuthenticatedClient | None = None,
    format_: str = None,
    offset: int = None,
    limit: int = None
):
    """Search for CRS reports using the universal search system."""
    from congressgov.services.core.search import search_async as universal_search

    return await universal_search(
        'crs-report',
        self,
        client=client,
        format_=format_,
        offset=offset,
        limit=limit
    )

AsyncCommittee ¶

AsyncCommittee(client: 'AuthenticatedClient | None' = None)

Bases: AsyncApiService

Committee service provides API access for fetching Congressional committee data.

Separation of Concerns: - This service class: Fetches data from API (get, search) - Committee model extensions: Operate on fetched data (expand, get_bills, etc.) - Committees model extensions: Query and filter collections

Source code in src/congressgov/services/async_api/committee.py

def __init__(self, client: 'AuthenticatedClient | None' = None) -> None:
    self.client = client
    self._last_result: 'CommitteeModel | None' = None

get `async` ¶

get(*, client: 'AuthenticatedClient | None' = None, chamber: str, committee_code: str, format_: str | None = None) -> 'CommitteeModel'

Get detailed information about a specific committee.

Source code in src/congressgov/services/async_api/committee.py

async def get(
    self,
    *,
    client: 'AuthenticatedClient | None' = None,
    chamber: str,
    committee_code: str,
    format_: str | None = None
) -> 'CommitteeModel':
    """Get detailed information about a specific committee."""
    # --- <VALIDATE PARAMETERS> ---
    validate_chamber(chamber)
    validate_committee_code(committee_code)

    resolved_client = await AsyncApiService._resolve_client(self, client)
    chamber_param = GetCommitteeChamberCommitteeCodeChamber(chamber.lower())

    response = await committee_details_async(
        client=resolved_client,
        chamber=chamber_param,
        committee_code=committee_code,
        format_=resolve_response_format(format_, GetCommitteeChamberCommitteeCodeFormat),
    )
    response_json = json.loads(response.content)

    api_envelope = ApiEnvelope.model_validate(response_json)
    result = CommitteeModel.model_validate(api_envelope.data)

    result.client = resolved_client
    self._last_result = result

    return result

get_by_congress `async` ¶

get_by_congress(*, client: 'AuthenticatedClient | None' = None, congress: int, chamber: str, committee_code: str, format_: str | None = None) -> 'CommitteeModel'

Get committee detail scoped to a specific Congress.

Source code in src/congressgov/services/async_api/committee.py

async def get_by_congress(
    self,
    *,
    client: 'AuthenticatedClient | None' = None,
    congress: int,
    chamber: str,
    committee_code: str,
    format_: str | None = None,
) -> 'CommitteeModel':
    """Get committee detail scoped to a specific Congress."""
    validate_congress(congress)
    validate_chamber(chamber)
    validate_committee_code(committee_code)

    resolved_client = await AsyncApiService._resolve_client(self, client)
    chamber_param = GetCommitteeCongressChamberCommitteeCodeChamber(chamber)
    response = await committee_by_congress_detailed(
        client=resolved_client,
        congress=congress,
        chamber=chamber_param,
        committee_code=committee_code,
        format_=resolve_response_format(
            format_, GetCommitteeCongressChamberCommitteeCodeFormat
        ),
    )
    response_json = json.loads(response.content)
    api_envelope = ApiEnvelope.model_validate(response_json)
    result = CommitteeModel.model_validate(api_envelope.data)
    result.client = resolved_client
    self._last_result = result
    return result

search `async` ¶

search(*, client: 'AuthenticatedClient | None' = None, congress: int | None = None, chamber: str | None = None, format_: str | None = None, offset: int | None = None, limit: int | None = None) -> 'CommitteesModel'

Search for committees using the universal search system.

Source code in src/congressgov/services/async_api/committee.py

async def search(
    self,
    *,
    client: 'AuthenticatedClient | None' = None,
    congress: int | None = None,
    chamber: str | None = None,
    format_: str | None = None,
    offset: int | None = None,
    limit: int | None = None
) -> 'CommitteesModel':
    """Search for committees using the universal search system."""
    from congressgov.services.core.search import search_async as universal_search

    return await universal_search(
        'committee',
        self,
        client=client,
        congress=congress,
        chamber=chamber,
        format_=format_,
        offset=offset,
        limit=limit
    )

AsyncCommitteeMeeting ¶

AsyncCommitteeMeeting(client: AuthenticatedClient | None = None)

Bases: AsyncApiService

CommitteeMeeting service provides API access for committee meeting data.

Source code in src/congressgov/services/async_api/committee_meeting.py

def __init__(self, client: AuthenticatedClient | None = None):
    self.client = client
    self._last_result = None

get `async` ¶

get(*, client: AuthenticatedClient | None = None, congress: int, chamber: str, event_id: str, format_: str = None) -> CommitteeMeetingModel

Get detailed information about a specific committee meeting.

Source code in src/congressgov/services/async_api/committee_meeting.py

async def get(
    self,
    *,
    client: AuthenticatedClient | None = None,
    congress: int,
    chamber: str,
    event_id: str,
    format_: str = None
) -> CommitteeMeetingModel:
    """Get detailed information about a specific committee meeting."""
    validate_congress(congress)
    validate_chamber(chamber)
    resolved_client = await AsyncApiService._resolve_client(self, client)
    response = await committee_meeting_detail_async(
        client=resolved_client,
        congress=congress,
        chamber=resolve_chamber(chamber, GetCommitteeMeetingCongressChamberEventIdChamber),
        event_id=event_id,
        format_=resolve_response_format(
            format_, GetCommitteeMeetingCongressChamberEventIdFormat
        ),
    )
    response_json = json.loads(response.content)
    api_envelope = ApiEnvelope.model_validate(response_json)
    raise_for_envelope_detail_response(
        response,
        response_json,
        api_envelope,
        resource_label="Committee meeting",
        identifiers={
            "congress": congress,
            "chamber": chamber,
            "event_id": event_id,
        },
    )
    result = CommitteeMeetingModel.model_validate(api_envelope.data)
    result.client = resolved_client
    self._last_result = result
    return result

search `async` ¶

search(*, client: AuthenticatedClient | None = None, congress: int = None, chamber: str = None, format_: str = None, offset: int = None, limit: int = None) -> CommitteeMeetingsModel

Search for committee meetings using the universal search system.

Source code in src/congressgov/services/async_api/committee_meeting.py

async def search(
    self,
    *,
    client: AuthenticatedClient | None = None,
    congress: int = None,
    chamber: str = None,
    format_: str = None,
    offset: int = None,
    limit: int = None
) -> CommitteeMeetingsModel:
    """Search for committee meetings using the universal search system."""
    from congressgov.services.core.search import search_async as universal_search

    return await universal_search(
        'committee-meeting',
        self,
        client=client,
        congress=congress,
        chamber=chamber,
        format_=format_,
        offset=offset,
        limit=limit
    )

AsyncCommitteePrint ¶

AsyncCommitteePrint(client: AuthenticatedClient | None = None)

Bases: AsyncApiService

CommitteePrint service provides API access for committee print data.

Source code in src/congressgov/services/async_api/committee_print.py

def __init__(self, client: AuthenticatedClient | None = None):
    self.client = client
    self._last_result = None

get `async` ¶

get(*, client: AuthenticatedClient | None = None, congress: int, jacket_number: int, chamber: str, format_: str = None) -> CommitteePrintModel

Get detailed information about a specific committee print.

Source code in src/congressgov/services/async_api/committee_print.py

async def get(
    self,
    *,
    client: AuthenticatedClient | None = None,
    congress: int,
    jacket_number: int,
    chamber: str,
    format_: str = None
) -> CommitteePrintModel:
    """Get detailed information about a specific committee print."""
    validate_congress(congress)
    validate_chamber(chamber)

    resolved_client = await AsyncApiService._resolve_client(self, client)
    response = await committee_print_detail_async(
        client=resolved_client,
        congress=congress,
        jacket_number=jacket_number,
        chamber=chamber,
        format_=format_
    )
    response_json = json.loads(response.content)
    api_envelope = ApiEnvelope.model_validate(response_json)
    result = CommitteePrintModel.model_validate(api_envelope.data)
    result.client = resolved_client
    self._last_result = result
    return result

get_text `async` ¶

get_text(*, client: AuthenticatedClient | None = None, congress: int, chamber: str, jacket_number: int, format_: str = None)

Return text versions for a committee print.

Source code in src/congressgov/services/async_api/committee_print.py

async def get_text(
    self,
    *,
    client: AuthenticatedClient | None = None,
    congress: int,
    chamber: str,
    jacket_number: int,
    format_: str = None,
):
    """Return text versions for a committee print."""
    validate_congress(congress)
    validate_chamber(chamber)

    resolved_client = await AsyncApiService._resolve_client(self, client)
    response = await committee_print_text_async(
        client=resolved_client,
        congress=congress,
        chamber=chamber,
        jacket_number=jacket_number,
        format_=format_,
    )
    response_json = json.loads(response.content)
    api_envelope = ApiEnvelope.model_validate(response_json)
    payload = response_json.get("data", api_envelope.data)
    if isinstance(payload, list):
        return [CommitteePrintTextModel.model_validate(item) for item in payload]
    return CommitteePrintTextModel.model_validate(payload)

search `async` ¶

search(*, client: AuthenticatedClient | None = None, congress: int = None, format_: str = None, offset: int = None, limit: int = None) -> CommitteePrintsModel

Search for committee prints using the universal search system.

Source code in src/congressgov/services/async_api/committee_print.py

async def search(
    self,
    *,
    client: AuthenticatedClient | None = None,
    congress: int = None,
    format_: str = None,
    offset: int = None,
    limit: int = None
) -> CommitteePrintsModel:
    """Search for committee prints using the universal search system."""
    from congressgov.services.core.search import search_async as universal_search

    return await universal_search(
        'committee-print',
        self,
        client=client,
        congress=congress,
        format_=format_,
        offset=offset,
        limit=limit
    )

AsyncCommitteeReport ¶

AsyncCommitteeReport(client: AuthenticatedClient | None = None)

Bases: AsyncApiService

CommitteeReport service provides API access for committee report data.

Source code in src/congressgov/services/async_api/committee_report.py

def __init__(self, client: AuthenticatedClient | None = None):
    self.client = client
    self._last_result = None

get `async` ¶

get(*, client: AuthenticatedClient | None = None, congress: int, report_type: str, report_number: int, format_: str = None) -> CommitteeReportModel

Get detailed information about a specific committee report.

Source code in src/congressgov/services/async_api/committee_report.py

async def get(
    self,
    *,
    client: AuthenticatedClient | None = None,
    congress: int,
    report_type: str,
    report_number: int,
    format_: str = None
) -> CommitteeReportModel:
    """Get detailed information about a specific committee report."""
    validate_congress(congress)

    resolved_client = await AsyncApiService._resolve_client(self, client)
    response = await committee_report_details_async(
        client=resolved_client,
        congress=congress,
        report_type=report_type,
        report_number=report_number,
        format_=format_
    )
    response_json = json.loads(response.content)
    api_envelope = ApiEnvelope.model_validate(response_json)
    result = CommitteeReportModel.model_validate(api_envelope.data)
    result.client = resolved_client
    self._last_result = result
    return result

get_text `async` ¶

get_text(*, client: AuthenticatedClient | None = None, congress: int, report_type: str, report_number: int, format_: str = None)

Return text versions for a committee report.

Source code in src/congressgov/services/async_api/committee_report.py

async def get_text(
    self,
    *,
    client: AuthenticatedClient | None = None,
    congress: int,
    report_type: str,
    report_number: int,
    format_: str = None,
):
    """Return text versions for a committee report."""
    validate_congress(congress)

    resolved_client = await AsyncApiService._resolve_client(self, client)
    response = await committee_report_id_text_async(
        client=resolved_client,
        congress=congress,
        report_type=report_type,
        report_number=report_number,
        format_=format_,
    )
    response_json = json.loads(response.content)
    api_envelope = ApiEnvelope.model_validate(response_json)
    payload = response_json.get("data", api_envelope.data)
    if isinstance(payload, list):
        return [CommitteeReportTextModel.model_validate(item) for item in payload]
    return CommitteeReportTextModel.model_validate(payload)

search `async` ¶

search(*, client: AuthenticatedClient | None = None, congress: int = None, report_type: str = None, format_: str = None, offset: int = None, limit: int = None) -> CommitteeReportsModel

Search for committee reports using the universal search system.

Source code in src/congressgov/services/async_api/committee_report.py

async def search(
    self,
    *,
    client: AuthenticatedClient | None = None,
    congress: int = None,
    report_type: str = None,
    format_: str = None,
    offset: int = None,
    limit: int = None
) -> CommitteeReportsModel:
    """Search for committee reports using the universal search system."""
    from congressgov.services.core.search import search_async as universal_search

    return await universal_search(
        'committee-report',
        self,
        client=client,
        congress=congress,
        report_type=report_type,
        format_=format_,
        offset=offset,
        limit=limit
    )

AsyncCongress ¶

AsyncCongress(client: AuthenticatedClient | None = None)

Bases: AsyncApiService

Congress service provides API access for Congressional session data.

Source code in src/congressgov/services/async_api/congress.py

def __init__(self, client: AuthenticatedClient | None = None):
    self.client = client
    self._last_result = None

current `async` ¶

current(*, client: AuthenticatedClient | None = None, format_: str = None) -> CongressModel

Get the current Congress session information.

Source code in src/congressgov/services/async_api/congress.py

async def current(
    self,
    *,
    client: AuthenticatedClient | None = None,
    format_: str = None
) -> CongressModel:
    """Get the current Congress session information."""
    resolved_client = await AsyncApiService._resolve_client(self, client)
    response = await congress_current_list_async(
        client=resolved_client,
        format_=resolve_response_format(format_, GetCongressCurrentFormat),
    )
    response_json = json.loads(response.content)
    api_envelope = ApiEnvelope.model_validate(response_json)
    result = CongressModel.model_validate(api_envelope.data)
    result.client = resolved_client
    return result

get `async` ¶

get(*, client: AuthenticatedClient | None = None, congress: int, format_: str = None) -> CongressModel

Get detailed information about a specific Congress session.

Source code in src/congressgov/services/async_api/congress.py

async def get(
    self,
    *,
    client: AuthenticatedClient | None = None,
    congress: int,
    format_: str = None
) -> CongressModel:
    """Get detailed information about a specific Congress session."""
    validate_congress(congress)
    resolved_client = await AsyncApiService._resolve_client(self, client)
    response = await congress_details_async(
        client=resolved_client,
        congress=congress,
        format_=resolve_response_format(format_, GetCongressCongressFormat),
    )
    response_json = json.loads(response.content)
    api_envelope = ApiEnvelope.model_validate(response_json)
    result = CongressModel.model_validate(api_envelope.data)
    _ensure_congress_number(result, congress)
    result.client = resolved_client
    self._last_result = result
    return result

search `async` ¶

search(*, client: AuthenticatedClient | None = None, format_: str = None, offset: int = None, limit: int = None) -> CongressesModel

List all Congress sessions.

Source code in src/congressgov/services/async_api/congress.py

async def search(
    self,
    *,
    client: AuthenticatedClient | None = None,
    format_: str = None,
    offset: int = None,
    limit: int = None
) -> CongressesModel:
    """List all Congress sessions."""
    resolved_client = await AsyncApiService._resolve_client(self, client)
    response = await congress_list_async(
        client=resolved_client,
        format_=resolve_response_format(format_, GetCongressFormat),
        offset=offset,
        limit=limit,
    )
    response_json = json.loads(response.content)
    api_envelope = ApiEnvelope.model_validate(response_json)
    result = CongressesModel.model_validate(api_envelope.data)
    result.client = resolved_client
    propagate_client_to_items(result, "congresses", resolved_client)
    return result

AsyncCongressionalRecord ¶

AsyncCongressionalRecord(client: AuthenticatedClient | None = None)

Bases: AsyncApiService

CongressionalRecord service provides API access for congressional record data.

Source code in src/congressgov/services/async_api/congressional_record.py

def __init__(self, client: AuthenticatedClient | None = None):
    self.client = client

search `async` ¶

search(*, client: AuthenticatedClient | None = None, year: int = None, month: int = None, day: int = None, format_: str = None, offset: int = None, limit: int = None)

Search for congressional records using the universal search system.

Source code in src/congressgov/services/async_api/congressional_record.py

async def search(
    self,
    *,
    client: AuthenticatedClient | None = None,
    year: int = None,
    month: int = None,
    day: int = None,
    format_: str = None,
    offset: int = None,
    limit: int = None
):
    """Search for congressional records using the universal search system."""
    from congressgov.services.core.search import search_async as universal_search

    return await universal_search(
        'congressional-record',
        self,
        client=client,
        year=year,
        month=month,
        day=day,
        format_=format_,
        offset=offset,
        limit=limit
    )

AsyncDailyCongressionalRecord ¶

AsyncDailyCongressionalRecord(client: AuthenticatedClient | None = None)

Bases: AsyncApiService

Async daily Congressional Record API.

Source code in src/congressgov/services/async_api/daily_congressional_record.py

def __init__(self, client: AuthenticatedClient | None = None):
    self.client = client
    self._last_result = None

AsyncHearing ¶

AsyncHearing(client: AuthenticatedClient | None = None)

Bases: AsyncApiService

Hearing service provides API access for Congressional hearing data.

Source code in src/congressgov/services/async_api/hearing.py

def __init__(self, client: AuthenticatedClient | None = None):
    self.client = client
    self._last_result = None

get `async` ¶

get(*, client: AuthenticatedClient | None = None, congress: int, chamber: str, jacket_number: int, format_: str = None) -> HearingModel

Get detailed information about a specific hearing.

Source code in src/congressgov/services/async_api/hearing.py

async def get(
    self,
    *,
    client: AuthenticatedClient | None = None,
    congress: int,
    chamber: str,
    jacket_number: int,
    format_: str = None
) -> HearingModel:
    """Get detailed information about a specific hearing."""
    resolved_client = await AsyncApiService._resolve_client(self, client)

    response = await hearing_detail_async(
        client=resolved_client,
        congress=congress,
        chamber=chamber,
        jacket_number=jacket_number,
        format_=format_,
    )
    response_json = json.loads(response.content)

    api_envelope = ApiEnvelope.model_validate(response_json)
    result = HearingModel.model_validate(api_envelope.data)

    result.client = resolved_client
    self._last_result = result

    return result

search `async` ¶

search(*, client: AuthenticatedClient | None = None, congress: int = None, chamber: str = None, format_: str = None, offset: int = None, limit: int = None) -> HearingsModel

Search for hearings using the universal search system.

Source code in src/congressgov/services/async_api/hearing.py

async def search(
    self,
    *,
    client: AuthenticatedClient | None = None,
    congress: int = None,
    chamber: str = None,
    format_: str = None,
    offset: int = None,
    limit: int = None
) -> HearingsModel:
    """Search for hearings using the universal search system."""
    from congressgov.services.core.search import search_async as universal_search

    return await universal_search(
        'hearing',
        self,
        client=client,
        congress=congress,
        chamber=chamber,
        format_=format_,
        offset=offset,
        limit=limit
    )

AsyncHouseCommunication ¶

AsyncHouseCommunication(client: AuthenticatedClient | None = None)

Bases: AsyncApiService

HouseCommunication service provides API access for House communication data.

Source code in src/congressgov/services/async_api/house_communication.py

def __init__(self, client: AuthenticatedClient | None = None):
    self.client = client
    self._last_result = None

get `async` ¶

get(*, client: AuthenticatedClient | None = None, congress: int, communication_type: str, communication_number: int, format_: str = None) -> HouseCommunicationModel

Get detailed information about a specific House communication.

Source code in src/congressgov/services/async_api/house_communication.py

async def get(
    self,
    *,
    client: AuthenticatedClient | None = None,
    congress: int,
    communication_type: str,
    communication_number: int,
    format_: str = None
) -> HouseCommunicationModel:
    """Get detailed information about a specific House communication."""
    resolved_client = await AsyncApiService._resolve_client(self, client)
    response = await house_communication_detail_async(
        client=resolved_client,
        congress=congress,
        communication_type=communication_type,
        communication_number=communication_number,
        format_=format_
    )
    response_json = json.loads(response.content)
    api_envelope = ApiEnvelope.model_validate(response_json)
    result = HouseCommunicationModel.model_validate(api_envelope.data)
    result.client = resolved_client
    self._last_result = result
    return result

search `async` ¶

search(*, client: AuthenticatedClient | None = None, congress: int = None, communication_type: str = None, format_: str = None, offset: int = None, limit: int = None) -> HouseCommunicationsModel

Search for House communications using the universal search system.

Source code in src/congressgov/services/async_api/house_communication.py

async def search(
    self,
    *,
    client: AuthenticatedClient | None = None,
    congress: int = None,
    communication_type: str = None,
    format_: str = None,
    offset: int = None,
    limit: int = None
) -> HouseCommunicationsModel:
    """Search for House communications using the universal search system."""
    from congressgov.services.core.search import search_async as universal_search

    return await universal_search(
        'house-communication',
        self,
        client=client,
        congress=congress,
        communication_type=communication_type,
        format_=format_,
        offset=offset,
        limit=limit
    )

AsyncHouseRequirement ¶

AsyncHouseRequirement(client: AuthenticatedClient | None = None)

Bases: AsyncApiService

HouseRequirement service provides API access for House requirement data.

Source code in src/congressgov/services/async_api/house_requirement.py

def __init__(self, client: AuthenticatedClient | None = None):
    self.client = client
    self._last_result = None

get `async` ¶

get(*, client: AuthenticatedClient | None = None, requirement_number: int, format_: str = None) -> HouseRequirementModel

Get detailed information about a specific House requirement.

Source code in src/congressgov/services/async_api/house_requirement.py

async def get(
    self,
    *,
    client: AuthenticatedClient | None = None,
    requirement_number: int,
    format_: str = None
) -> HouseRequirementModel:
    """Get detailed information about a specific House requirement."""
    resolved_client = await AsyncApiService._resolve_client(self, client)
    response = await house_requirement_detail_async(
        client=resolved_client,
        requirement_number=requirement_number,
        format_=format_
    )
    response_json = json.loads(response.content)
    api_envelope = ApiEnvelope.model_validate(response_json)
    result = HouseRequirementModel.model_validate(api_envelope.data)
    result.client = resolved_client
    self._last_result = result
    return result

search `async` ¶

search(*, client: AuthenticatedClient | None = None, format_: str = None, offset: int = None, limit: int = None) -> HouseRequirementsModel

Search for House requirements using the universal search system.

Source code in src/congressgov/services/async_api/house_requirement.py

async def search(
    self,
    *,
    client: AuthenticatedClient | None = None,
    format_: str = None,
    offset: int = None,
    limit: int = None
) -> HouseRequirementsModel:
    """Search for House requirements using the universal search system."""
    from congressgov.services.core.search import search_async as universal_search

    return await universal_search(
        'house-requirement',
        self,
        client=client,
        format_=format_,
        offset=offset,
        limit=limit
    )