Module adcp.types.guards

Type guards for ADCP discriminated union responses.

ADCP 3.0 responses use atomic semantics: a response contains EITHER success data OR errors, never both. These type guards enable static type narrowing with mypy and runtime branch detection.

Usage

from adcp.types.guards import is_success, is_error

response = result.data # CreateMediaBuyResponse (union type)

if is_success(response): # mypy knows this is CreateMediaBuySuccessResponse print(response.media_buy_id) else: # error branch print(response.errors)

Generic guards work with any ADCP response union: from adcp.types.guards import is_adcp_error, is_adcp_success

if is_adcp_error(response):
    print(response.errors)

Functions

def is_activate_signal_error(response: ActivateSignalResponse) ‑> TypeGuard[adcp.types.generated_poc.signals.activate_signal_response.ActivateSignalResponse2]

Expand source code

def is_activate_signal_error(
    response: ActivateSignalResponse,
) -> TypeGuard[ActivateSignalErrorResponse]:
    """Check if an ActivateSignalResponse is an error."""
    return is_adcp_error(response)

Check if an ActivateSignalResponse is an error.

def is_activate_signal_success(response: ActivateSignalResponse) ‑> TypeGuard[adcp.types.generated_poc.signals.activate_signal_response.ActivateSignalResponse1]

Expand source code

def is_activate_signal_success(
    response: ActivateSignalResponse,
) -> TypeGuard[ActivateSignalSuccessResponse]:
    """Check if an ActivateSignalResponse is a success."""
    return not is_adcp_error(response)

Check if an ActivateSignalResponse is a success.

def is_adcp_error(response: Any) ‑> bool

Expand source code

def is_adcp_error(response: Any) -> bool:
    """Check if an ADCP response is an error response.

    Works with any ADCP response union type. Error responses
    have a non-empty ``errors`` field.

    Args:
        response: Any ADCP response object (success or error variant).

    Returns:
        True if the response contains errors.
    """
    errors = getattr(response, "errors", None)
    if errors is None:
        return False
    if isinstance(errors, list):
        return len(errors) > 0
    return True

Check if an ADCP response is an error response.

Works with any ADCP response union type. Error responses have a non-empty errors field.

Args

response

Any ADCP response object (success or error variant).

Returns

True if the response contains errors.

def is_adcp_success(response: Any) ‑> bool

Expand source code

def is_adcp_success(response: Any) -> bool:
    """Check if an ADCP response is a success response.

    Works with any ADCP response union type. Success responses
    do not have an ``errors`` field (or it is None/empty).

    Args:
        response: Any ADCP response object (success or error variant).

    Returns:
        True if the response is a success (no errors).
    """
    return not is_adcp_error(response)

Check if an ADCP response is a success response.

Works with any ADCP response union type. Success responses do not have an errors field (or it is None/empty).

Args

response

Any ADCP response object (success or error variant).

Returns

True if the response is a success (no errors).

def is_build_creative_error(response: BuildCreativeResponse) ‑> TypeGuard[adcp.types.generated_poc.media_buy.build_creative_response.BuildCreativeResponse2]

Expand source code

def is_build_creative_error(
    response: BuildCreativeResponse,
) -> TypeGuard[BuildCreativeErrorResponse]:
    """Check if a BuildCreativeResponse is an error."""
    if is_build_creative_submitted(response):
        return False
    return is_adcp_error(response)

Check if a BuildCreativeResponse is an error.

def is_build_creative_submitted(response: BuildCreativeResponse) ‑> TypeGuard[adcp.types.generated_poc.media_buy.build_creative_response.BuildCreativeResponse6]

Expand source code

def is_build_creative_submitted(
    response: BuildCreativeResponse,
) -> TypeGuard[BuildCreativeSubmittedResponse]:
    """Check if a BuildCreativeResponse is the async submitted envelope."""
    return getattr(response, "status", None) == "submitted" and hasattr(response, "task_id")

Check if a BuildCreativeResponse is the async submitted envelope.

def is_build_creative_success(response: BuildCreativeResponse) ‑> TypeGuard[adcp.types.generated_poc.media_buy.build_creative_response.BuildCreativeResponse1 | adcp.types.generated_poc.media_buy.build_creative_response.BuildCreativeResponse3 | adcp.types.generated_poc.media_buy.build_creative_response.BuildCreativeResponse4 | adcp.types.generated_poc.media_buy.build_creative_response.BuildCreativeResponse5]

Expand source code

def is_build_creative_success(
    response: BuildCreativeResponse,
) -> TypeGuard[BuildCreativeSuccessBranches]:
    """Check if a BuildCreativeResponse is a synchronous success."""
    if is_build_creative_submitted(response):
        return False
    return not is_adcp_error(response)

Check if a BuildCreativeResponse is a synchronous success.

def is_calibrate_content_success(response: CalibrateContentSuccessResponse | CalibrateContentErrorResponse) ‑> TypeGuard[adcp.types.generated_poc.content_standards.calibrate_content_response.CalibrateContentResponse1]

Expand source code

def is_calibrate_content_success(
    response: CalibrateContentSuccessResponse | CalibrateContentErrorResponse,
) -> TypeGuard[CalibrateContentSuccessResponse]:
    """Check if a CalibrateContentResponse is a success."""
    return not is_adcp_error(response)

Check if a CalibrateContentResponse is a success.

def is_create_media_buy_error(response: CreateMediaBuyResponse) ‑> TypeGuard[adcp.types.generated_poc.media_buy.create_media_buy_response.CreateMediaBuyResponse2]

Expand source code

def is_create_media_buy_error(
    response: CreateMediaBuyResponse,
) -> TypeGuard[CreateMediaBuyErrorResponse]:
    """Check if a CreateMediaBuyResponse is an error.

    Returns False for the submitted (async) envelope, even if it carries
    advisory (non-blocking) errors. Use ``is_create_media_buy_submitted``
    for that branch.
    """
    if is_create_media_buy_submitted(response):
        return False
    return is_adcp_error(response)

Check if a CreateMediaBuyResponse is an error.

Returns False for the submitted (async) envelope, even if it carries advisory (non-blocking) errors. Use is_create_media_buy_submitted() for that branch.

def is_create_media_buy_submitted(response: CreateMediaBuyResponse) ‑> TypeGuard[adcp.types.generated_poc.media_buy.create_media_buy_response.CreateMediaBuyResponse3]

Expand source code

def is_create_media_buy_submitted(
    response: CreateMediaBuyResponse,
) -> TypeGuard[CreateMediaBuySubmittedResponse]:
    """Check if a CreateMediaBuyResponse is the async submitted envelope.

    The submitted branch carries ``status == 'submitted'`` and a ``task_id``
    the buyer uses to poll ``tasks/get`` (or correlate with push-notification
    callbacks). It is neither a synchronous success nor a terminal error.
    """
    return getattr(response, "status", None) == "submitted" and hasattr(response, "task_id")

Check if a CreateMediaBuyResponse is the async submitted envelope.

The submitted branch carries status == 'submitted' and a task_id the buyer uses to poll tasks/get (or correlate with push-notification callbacks). It is neither a synchronous success nor a terminal error.

def is_create_media_buy_success(response: CreateMediaBuyResponse) ‑> TypeGuard[adcp.types.generated_poc.media_buy.create_media_buy_response.CreateMediaBuyResponse1]

Expand source code

def is_create_media_buy_success(
    response: CreateMediaBuyResponse,
) -> TypeGuard[CreateMediaBuySuccessResponse]:
    """Check if a CreateMediaBuyResponse is a synchronous success.

    Returns False for the submitted (async) envelope — use
    ``is_create_media_buy_submitted`` for that branch.
    """
    if is_create_media_buy_submitted(response):
        return False
    return not is_adcp_error(response)

Check if a CreateMediaBuyResponse is a synchronous success.

Returns False for the submitted (async) envelope — use is_create_media_buy_submitted() for that branch.

def is_get_account_financials_error(response: GetAccountFinancialsSuccessResponse | GetAccountFinancialsErrorResponse) ‑> TypeGuard[adcp.types.generated_poc.account.get_account_financials_response.GetAccountFinancialsResponse2]

Expand source code

def is_get_account_financials_error(
    response: GetAccountFinancialsSuccessResponse | GetAccountFinancialsErrorResponse,
) -> TypeGuard[GetAccountFinancialsErrorResponse]:
    """Check if a GetAccountFinancialsResponse is an error."""
    return is_adcp_error(response)

Check if a GetAccountFinancialsResponse is an error.

def is_get_account_financials_success(response: GetAccountFinancialsSuccessResponse | GetAccountFinancialsErrorResponse) ‑> TypeGuard[adcp.types.generated_poc.account.get_account_financials_response.GetAccountFinancialsResponse1]

Expand source code

def is_get_account_financials_success(
    response: GetAccountFinancialsSuccessResponse | GetAccountFinancialsErrorResponse,
) -> TypeGuard[GetAccountFinancialsSuccessResponse]:
    """Check if a GetAccountFinancialsResponse is a success."""
    return not is_adcp_error(response)

Check if a GetAccountFinancialsResponse is a success.

def is_get_creative_features_success(response: GetCreativeFeaturesSuccessResponse | GetCreativeFeaturesErrorResponse) ‑> TypeGuard[adcp.types.generated_poc.creative.get_creative_features_response.GetCreativeFeaturesResponse1]

Expand source code

def is_get_creative_features_success(
    response: GetCreativeFeaturesSuccessResponse | GetCreativeFeaturesErrorResponse,
) -> TypeGuard[GetCreativeFeaturesSuccessResponse]:
    """Check if a GetCreativeFeaturesResponse is a success."""
    return not is_adcp_error(response)

Check if a GetCreativeFeaturesResponse is a success.

def is_log_event_error(response: LogEventResponse) ‑> TypeGuard[adcp.types.generated_poc.media_buy.log_event_response.LogEventResponse2]

Expand source code

def is_log_event_error(
    response: LogEventResponse,
) -> TypeGuard[LogEventErrorResponse]:
    """Check if a LogEventResponse is an error."""
    return is_adcp_error(response)

Check if a LogEventResponse is an error.

def is_log_event_success(response: LogEventResponse) ‑> TypeGuard[adcp.types.generated_poc.media_buy.log_event_response.LogEventResponse1]

Expand source code

def is_log_event_success(
    response: LogEventResponse,
) -> TypeGuard[LogEventSuccessResponse]:
    """Check if a LogEventResponse is a success."""
    return not is_adcp_error(response)

Check if a LogEventResponse is a success.

def is_performance_feedback_error(response: ProvidePerformanceFeedbackSuccessResponse | ProvidePerformanceFeedbackErrorResponse) ‑> TypeGuard[adcp.types.generated_poc.media_buy.provide_performance_feedback_response.ProvidePerformanceFeedbackResponse2]

Expand source code

def is_performance_feedback_error(
    response: ProvidePerformanceFeedbackSuccessResponse | ProvidePerformanceFeedbackErrorResponse,
) -> TypeGuard[ProvidePerformanceFeedbackErrorResponse]:
    """Check if a ProvidePerformanceFeedbackResponse is an error."""
    return is_adcp_error(response)

Check if a ProvidePerformanceFeedbackResponse is an error.

def is_performance_feedback_success(response: ProvidePerformanceFeedbackSuccessResponse | ProvidePerformanceFeedbackErrorResponse) ‑> TypeGuard[adcp.types.generated_poc.media_buy.provide_performance_feedback_response.ProvidePerformanceFeedbackResponse1]

Expand source code

def is_performance_feedback_success(
    response: ProvidePerformanceFeedbackSuccessResponse | ProvidePerformanceFeedbackErrorResponse,
) -> TypeGuard[ProvidePerformanceFeedbackSuccessResponse]:
    """Check if a ProvidePerformanceFeedbackResponse is a success."""
    return not is_adcp_error(response)

Check if a ProvidePerformanceFeedbackResponse is a success.

def is_sync_accounts_error(response: SyncAccountsResponse) ‑> TypeGuard[adcp.types.generated_poc.account.sync_accounts_response.SyncAccountsResponse2]

Expand source code

def is_sync_accounts_error(
    response: SyncAccountsResponse,
) -> TypeGuard[SyncAccountsErrorResponse]:
    """Check if a SyncAccountsResponse is an error."""
    return is_adcp_error(response)

Check if a SyncAccountsResponse is an error.

def is_sync_accounts_success(response: SyncAccountsResponse) ‑> TypeGuard[adcp.types.generated_poc.account.sync_accounts_response.SyncAccountsResponse1]

Expand source code

def is_sync_accounts_success(
    response: SyncAccountsResponse,
) -> TypeGuard[SyncAccountsSuccessResponse]:
    """Check if a SyncAccountsResponse is a success."""
    return not is_adcp_error(response)

Check if a SyncAccountsResponse is a success.

def is_sync_catalogs_error(response: SyncCatalogsResponse) ‑> TypeGuard[adcp.types.generated_poc.media_buy.sync_catalogs_response.SyncCatalogsResponse2]

Expand source code

def is_sync_catalogs_error(
    response: SyncCatalogsResponse,
) -> TypeGuard[SyncCatalogsErrorResponse]:
    """Check if a SyncCatalogsResponse is an error."""
    if is_sync_catalogs_submitted(response):
        return False
    return is_adcp_error(response)

Check if a SyncCatalogsResponse is an error.

def is_sync_catalogs_submitted(response: SyncCatalogsResponse) ‑> TypeGuard[adcp.types.generated_poc.media_buy.sync_catalogs_response.SyncCatalogsResponse3]

Expand source code

def is_sync_catalogs_submitted(
    response: SyncCatalogsResponse,
) -> TypeGuard[SyncCatalogsSubmittedResponse]:
    """Check if a SyncCatalogsResponse is the async submitted envelope."""
    return getattr(response, "status", None) == "submitted" and hasattr(response, "task_id")

Check if a SyncCatalogsResponse is the async submitted envelope.

def is_sync_catalogs_success(response: SyncCatalogsResponse) ‑> TypeGuard[adcp.types.generated_poc.media_buy.sync_catalogs_response.SyncCatalogsResponse1]

Expand source code

def is_sync_catalogs_success(
    response: SyncCatalogsResponse,
) -> TypeGuard[SyncCatalogsSuccessResponse]:
    """Check if a SyncCatalogsResponse is a synchronous success."""
    if is_sync_catalogs_submitted(response):
        return False
    return not is_adcp_error(response)

Check if a SyncCatalogsResponse is a synchronous success.

def is_sync_creatives_error(response: SyncCreativesResponse) ‑> TypeGuard[adcp.types.generated_poc.creative.sync_creatives_response.SyncCreativesResponse2]

Expand source code

def is_sync_creatives_error(
    response: SyncCreativesResponse,
) -> TypeGuard[SyncCreativesErrorResponse]:
    """Check if a SyncCreativesResponse is an error."""
    if is_sync_creatives_submitted(response):
        return False
    return is_adcp_error(response)

Check if a SyncCreativesResponse is an error.

def is_sync_creatives_submitted(response: SyncCreativesResponse) ‑> TypeGuard[adcp.types.generated_poc.creative.sync_creatives_response.SyncCreativesResponse3]

Expand source code

def is_sync_creatives_submitted(
    response: SyncCreativesResponse,
) -> TypeGuard[SyncCreativesSubmittedResponse]:
    """Check if a SyncCreativesResponse is the async submitted envelope."""
    return getattr(response, "status", None) == "submitted" and hasattr(response, "task_id")

Check if a SyncCreativesResponse is the async submitted envelope.

def is_sync_creatives_success(response: SyncCreativesResponse) ‑> TypeGuard[adcp.types.generated_poc.creative.sync_creatives_response.SyncCreativesResponse1]

Expand source code

def is_sync_creatives_success(
    response: SyncCreativesResponse,
) -> TypeGuard[SyncCreativesSuccessResponse]:
    """Check if a SyncCreativesResponse is a synchronous success."""
    if is_sync_creatives_submitted(response):
        return False
    return not is_adcp_error(response)

Check if a SyncCreativesResponse is a synchronous success.

def is_update_media_buy_error(response: UpdateMediaBuyResponse) ‑> TypeGuard[adcp.types.generated_poc.media_buy.update_media_buy_response.UpdateMediaBuyResponse2]

Expand source code

def is_update_media_buy_error(
    response: UpdateMediaBuyResponse,
) -> TypeGuard[UpdateMediaBuyErrorResponse]:
    """Check if an UpdateMediaBuyResponse is an error.

    Returns False for the submitted (async) envelope, even if it carries
    advisory (non-blocking) errors. Use ``is_update_media_buy_submitted``
    for that branch.
    """
    if is_update_media_buy_submitted(response):
        return False
    return is_adcp_error(response)

Check if an UpdateMediaBuyResponse is an error.

Returns False for the submitted (async) envelope, even if it carries advisory (non-blocking) errors. Use is_update_media_buy_submitted() for that branch.

def is_update_media_buy_submitted(response: UpdateMediaBuyResponse) ‑> TypeGuard[adcp.types.generated_poc.media_buy.update_media_buy_response.UpdateMediaBuyResponse3]

Expand source code

def is_update_media_buy_submitted(
    response: UpdateMediaBuyResponse,
) -> TypeGuard[UpdateMediaBuySubmittedResponse]:
    """Check if an UpdateMediaBuyResponse is the async submitted envelope.

    The submitted branch carries ``status == 'submitted'`` and a ``task_id``.
    It is neither a synchronous success nor a terminal error.
    """
    return getattr(response, "status", None) == "submitted" and hasattr(response, "task_id")

Check if an UpdateMediaBuyResponse is the async submitted envelope.

The submitted branch carries status == 'submitted' and a task_id. It is neither a synchronous success nor a terminal error.

def is_update_media_buy_success(response: UpdateMediaBuyResponse) ‑> TypeGuard[adcp.types.generated_poc.media_buy.update_media_buy_response.UpdateMediaBuyResponse1]

Expand source code

def is_update_media_buy_success(
    response: UpdateMediaBuyResponse,
) -> TypeGuard[UpdateMediaBuySuccessResponse]:
    """Check if an UpdateMediaBuyResponse is a synchronous success.

    Returns False for the submitted (async) envelope — use
    ``is_update_media_buy_submitted`` for that branch.
    """
    if is_update_media_buy_submitted(response):
        return False
    return not is_adcp_error(response)

Check if an UpdateMediaBuyResponse is a synchronous success.

Returns False for the submitted (async) envelope — use is_update_media_buy_submitted() for that branch.

def is_validate_content_delivery_success(response: ValidateContentDeliverySuccessResponse | ValidateContentDeliveryErrorResponse) ‑> TypeGuard[adcp.types.generated_poc.content_standards.validate_content_delivery_response.ValidateContentDeliveryResponse1]

Expand source code

def is_validate_content_delivery_success(
    response: ValidateContentDeliverySuccessResponse | ValidateContentDeliveryErrorResponse,
) -> TypeGuard[ValidateContentDeliverySuccessResponse]:
    """Check if a ValidateContentDeliveryResponse is a success."""
    return not is_adcp_error(response)

Check if a ValidateContentDeliveryResponse is a success.