API Reference

Top-level Package

Top-level public API for PipeBridge.

This module intentionally exposes a hybrid public surface:

  • the high-level PipeBridge facade for most consumers

  • the core services for advanced composition scenarios

  • the main request/config objects required by upload, update, and move flows

class pipebridge.PipeBridge(token, base_url, pipe_schema_cache_ttl_seconds=300, transport_config=None)[source]

Bases: object

Root public facade of the Pipefy SDK.

This class wires the HTTP client and all domain facades, exposing a cohesive and stable API surface for cards, phases, pipes, and files.

Parameters:
class pipebridge.PipefyHttpClient(auth_key, base_url, transport_config=None)[source]

Bases: object

Handles HTTP communication with Pipefy GraphQL API.

This class is responsible for: - Sending HTTP requests to Pipefy - Handling transport-level errors - Validating response structure - Mapping API errors to domain exceptions

It does NOT: - Apply business rules - Transform domain data

Parameters:

  • auth_key (str) – str = API authentication token

  • base_url (str) – str = Pipefy GraphQL endpoint

  • transport_config (TransportConfig | None)

Raises:
Example:
>>> client = PipefyHttpClient(auth_key="valid_token", base_url="www.pipebridge.example.com.br")
>>> isinstance(client, PipefyHttpClient)
True
sendRequest(query, variables=None, timeout=30)[source]

Sends a GraphQL request to Pipefy API.

Handles: - Payload construction - HTTP request execution - JSON parsing - Response validation - Error mapping

Parameters:

  • timeout (int)

  • query (str) – str = GraphQL query string

  • variables (Dict[str, Any] | None) – dict | None = Optional GraphQL variables

Returns:

dict = Raw API response

Raises:
Example:
>>> client = PipefyHttpClient(auth_key="valid_token", base_url="www.pipebridge.example.com.br")
>>> query = "{ me { id } }"
>>> isinstance(query, str)
True
Return type:

Dict[str, Any]

property transport_config: TransportConfig

Expose the effective transport configuration bound to this client.

class pipebridge.TransportConfig(timeout=30, verify_ssl=True, ca_bundle_path=None, max_retries=0, retry_delay_seconds=0.0, retry_backoff_multiplier=1.0)[source]

Bases: object

Transport-level configuration for HTTP communication with Pipefy.

Defaults preserve the current behavior while allowing TLS and timeout customization in a backward-compatible way.

Parameters:

  • timeout (int)

  • verify_ssl (bool)

  • ca_bundle_path (str | None)

  • max_retries (int)

  • retry_delay_seconds (float)

  • retry_backoff_multiplier (float)

timeout: int = 30
verify_ssl: bool = True
ca_bundle_path: str | None = None
max_retries: int = 0
retry_delay_seconds: float = 0.0
retry_backoff_multiplier: float = 1.0
resolveVerifyValue()[source]

Resolve the value expected by requests.post(…, verify=…).

Return type:

bool | str

class pipebridge.CardService(client, pipe_schema_cache_ttl_seconds=300)[source]

Bases: object

Service responsible for card operations in Pipefy.

This service provides a complete abstraction over Pipefy GraphQL for managing cards, including creation, retrieval, updates, and movement across phases.

param client:

PipefyHttpClient = HTTP client instance

example:
>>> service = CardService(PipefyHttpClient(auth_key="token", base_url="https://api.pipefy.com/graphql"))
>>> isinstance(service, CardService)
True
Parameters:
invalidatePipeSchemaCache(pipe_id=None)[source]

Invalidate the in-memory pipe schema cache.

Parameters:

pipe_id (str | None) – str | None = Specific pipe identifier. When omitted, the entire schema cache is invalidated.

Returns:

None

Return type:

None

getPipeSchemaCacheStats()[source]

Retrieve aggregate statistics from the pipe schema cache.

Returns:

dict = Cache statistics

Return type:

Dict[str, Any]

getPipeSchemaCacheEntryInfo(pipe_id)[source]

Retrieve metadata about a specific cached pipe schema entry.

Parameters:

pipe_id (str) – str = Pipe identifier

Returns:

dict | None = Cache entry metadata

Return type:

Dict[str, Any] | None

createCard(pipe_id, title, fields=None)[source]

Create a new card in a pipe.

This method creates a card in the specified pipe and optionally sets initial field values.

Parameters:

  • pipe_id (str) – str = Pipe identifier

  • title (str) – str = Card title

  • fields (Dict[str, Any] | None) – dict[str, Any] | None = Field values mapped by field_id

Returns:

dict = Raw API response containing created card metadata

Raises:
Example:
>>> service = CardService(PipefyHttpClient(auth_key="token", base_url="https://api.pipefy.com/graphql"))
>>> callable(service.createCard)
True
Return type:

Dict[str, Any]

createCardSafely(pipe_id, title, fields=None)[source]

Create a new card after validating input against the pipe start form schema.

Parameters:

  • pipe_id (str) – str = Pipe identifier

  • title (str) – str = Card title

  • fields (Dict[str, Any] | None) – dict[str, Any] | None = Start form field values

Returns:

dict = Raw API response containing created card metadata

Raises:
Return type:

Dict[str, Any]

getRawCard(card_id, query_body)[source]

Execute a raw GraphQL query for a card.

This method allows full control over the GraphQL query structure.

Parameters:

  • card_id (str) – str = Card identifier

  • query_body (str) – str = GraphQL body inside card query

Returns:

dict = Raw API response

Raises:
Example:
>>> service = CardService(PipefyHttpClient(auth_key="token", base_url="https://api.pipefy.com/graphql"))
>>> callable(service.getRawCard)
True
Return type:

Dict[str, Any]

getCardStructured(card_id, query_body=None)[source]

Retrieve structured card data.

This method abstracts GraphQL complexity by providing a default query that includes commonly used fields such as phases, fields, labels, assignees, and attachments.

Parameters:

  • card_id (str) – str = Card identifier

  • query_body (str | None) – str | None = Custom GraphQL body override

Returns:

dict = Structured card data

Raises:
Example:
>>> service = CardService(PipefyHttpClient(auth_key="token", base_url="https://api.pipefy.com/graphql"))
>>> callable(service.getCardStructured)
True
Return type:

Dict[str, Any]

getCardPipeField(card_id, field_id)[source]

Retrieve schema metadata for a card field anywhere in the parent pipe.

This helper is schema-oriented. It does not depend on whether the field currently appears in card.fields.

Parameters:

  • card_id (str) – str = Card identifier

  • field_id (str) – str = Field identifier

Returns:

PhaseField | None = Matching schema field when found

Return type:

PhaseField | None

getCardCurrentPhaseField(card_id, field_id)[source]

Retrieve schema metadata for a field in the card current phase.

This helper is schema-oriented. It does not depend on whether the field currently appears in card.fields.

Parameters:

  • card_id (str) – str = Card identifier

  • field_id (str) – str = Field identifier

Returns:

PhaseField | None = Matching current-phase field when found

Return type:

PhaseField | None

listCardAttachments(card_id)[source]

Retrieve all attachments exposed by the Pipefy card attachments surface.

This helper is more robust than reading attachment URLs from card.fields because it also works for attachment fields that are not materialized in the current card field payload.

Parameters:

card_id (str) – str = Card identifier

Returns:

list[dict[str, Any]] = Attachment payload items

Return type:

List[Dict[str, Any]]

listCardAttachmentsByField(card_id, field_id)[source]

Retrieve attachments associated with a specific field in a card.

Parameters:

  • card_id (str) – str = Card identifier

  • field_id (str) – str = Field identifier

Returns:

list[dict[str, Any]] = Attachment payload items for the field

Return type:

List[Dict[str, Any]]

moveCardToPhase(card_id, phase_id)[source]

Move a card to another phase in Pipefy.

This method executes a GraphQL mutation to move a card from its current phase to a destination phase.

Parameters:

  • card_id (str) – str = Unique identifier of the card to be moved

  • phase_id (str) – str = Destination phase ID

Returns:

dict = Raw API response containing updated card reference

Raises:
Example:
>>> service = CardService(PipefyHttpClient(auth_key="token", base_url="https://api.pipefy.com/graphql"))
>>> callable(service.moveCardToPhase)
True
Return type:

Dict[str, Any]

moveCardToPhaseSafely(card_id, destination_phase_id, expected_current_phase_id=None, config=None, extra_rules=None)[source]

Move a card to another phase after validating transition readiness.

This safe flow can validate destination required fields before executing the low-level move mutation.

Parameters:

  • card_id (str) – str = Card identifier

  • destination_phase_id (str) – str = Destination phase identifier

  • expected_current_phase_id (str | None) – str | None = Optional expected source phase identifier

  • config (CardMoveConfig | None) – CardMoveConfig | None = Optional flow configuration

  • extra_rules (List[BaseRule] | None) – list[BaseRule] | None = Optional additional rules

Returns:

dict = Raw API move response

Raises:
Return type:

Dict[str, Any]

searchCards(pipe_id, title)[source]

Search for cards in a pipe by title.

This method executes a GraphQL query to find cards whose titles match the provided search term.

Parameters:

  • pipe_id (str) – str = Pipe ID where the search will be performed

  • title (str) – str = Title or partial title used as search filter

Returns:

list[dict] = List of card nodes containing basic information (id, title)

Raises:
Example:
>>> service = CardService(PipefyHttpClient(auth_key="token", base_url="https://api.pipefy.com/graphql"))
>>> callable(service.searchCards)
True
Return type:

List[Dict[str, Any]]

relateCards(parent_id, child_id, source_id)[source]

Create a relationship between two cards in Pipefy.

This method executes a GraphQL mutation to establish a relationship between a parent card and a child card using a Pipe relation.

Parameters:

  • parent_id (str) – str = Parent card ID (source of the relationship)

  • child_id (str) – str = Child card ID (target of the relationship)

  • source_id (str) – str = Pipe relation ID used to define the relationship

Returns:

dict = Raw API response containing the created relation ID

Raises:
Example:
>>> service = CardService(PipefyHttpClient(auth_key="token", base_url="https://api.pipefy.com/graphql"))
>>> callable(service.relateCards)
True
Return type:

Dict[str, Any]

listAllCards(pipe_id, attributes=None, filters=None, only_phase_name=None)[source]

Retrieve all cards from a pipe using cursor-based pagination.

This method iteratively fetches all cards from a given pipe using GraphQL pagination. It supports optional filtering and phase-based restriction.

Parameters:

  • pipe_id (str) – str = Pipe ID from which cards will be retrieved

  • attributes (List[str] | None) – list[str] | None = List of attributes to include in the query. Defaults to [“id”, “title”]

  • filters (Dict[str, Any] | None) – dict | None = GraphQL filter object to refine results

  • only_phase_name (str | None) – str | None = If provided, only cards belonging to this phase name will be returned

Returns:

list[dict] = List of card nodes matching the query criteria

Raises:
Example:
>>> service = CardService(PipefyHttpClient(auth_key="token", base_url="https://api.pipefy.com/graphql"))
>>> callable(service.listAllCards)
True
Return type:

List[Dict[str, Any]]

listCardsFromPhase(phase_id, limit=50)[source]

Retrieve all cards from a specific phase using cursor-based pagination.

This method iteratively fetches cards from a given phase, handling pagination internally until all cards are retrieved.

Parameters:

  • phase_id (str) – str = Phase ID from which cards will be retrieved

  • limit (int) – int = Number of cards per request (page size)

Returns:

list[dict] = List of card nodes containing basic information (id, title)

Raises:
Example:
>>> service = CardService(PipefyHttpClient(auth_key="token", base_url="https://api.pipefy.com/graphql"))
>>> callable(service.listCardsFromPhase)
True
Return type:

List[Dict[str, Any]]

deleteCard(card_id)[source]

Delete a card from Pipefy.

This method executes a GraphQL mutation to permanently remove a card identified by its ID.

Parameters:

card_id (str) – str = Unique identifier of the card to be deleted

Returns:

dict = Raw API response indicating operation success

Raises:
Example:
>>> service = CardService(PipefyHttpClient(auth_key="token", base_url="https://api.pipefy.com/graphql"))
>>> callable(service.deleteCard)
True
Return type:

Dict[str, Any]

updateField(card_id, field_id, value, config=None, extra_rules=None, extra_handlers=None)[source]

Update a single field of a card.

This is a convenience wrapper over updateFields.

Parameters:

  • card_id (str) – str = Card identifier

  • field_id (str) – str = Field identifier

  • value (str) – str = Field value

  • config (CardUpdateConfig | None)

  • extra_rules (List[BaseRule] | None)

  • extra_handlers (Dict[str, BaseCardFieldUpdateHandler] | None)

Returns:

dict = Raw API response

Raises:
Example:
>>> service = CardService(PipefyHttpClient(auth_key="token", base_url="https://api.pipefy.com/graphql"))
>>> callable(service.updateField)
True
Return type:

Dict[str, Any]

updateFields(card_id, fields, expected_phase_id=None, config=None, extra_rules=None, extra_handlers=None)[source]

Updates multiple fields of a card using Pipefy API.

This method performs:

  • Phase validation (optional)

  • Field validation (required, options, existence)

  • Dynamic handling for attachments vs standard fields

  • Execution of GraphQL mutations per field

Parameters:

  • card_id (str) – str = Card identifier

  • fields (Dict[str, Any]) – dict[str, Any] = Mapping of field_id to new values

  • expected_phase_id (str | None) – Optional[str] = Expected phase ID for validation

  • config (CardUpdateConfig | None) – Optional[CardUpdateConfig] = Optional flow configuration

  • extra_rules (List[BaseRule] | None) – Optional[list[BaseRule]] = Optional custom validation rules

  • extra_handlers (Dict[str, BaseCardFieldUpdateHandler] | None)

Returns:

dict = Aggregated API responses per field

Raises:
Example:
>>> service = CardService(PipefyHttpClient(auth_key="token", base_url="https://api.pipefy.com/graphql"))
>>> callable(service._buildAttachmentMutation)
True
Return type:

Dict[str, Any]

updateAssigneeIds(card_id, assignee_ids)[source]

Update the assignees of a card.

This method executes a GraphQL mutation to assign one or more users to a card.

Parameters:

  • card_id (str) – str = Card ID to be updated

  • assignee_ids (List[str]) – list[str] = List of user IDs to assign to the card

Returns:

dict = Raw API response containing updated card data

Raises:
Example:
>>> service = CardService(PipefyHttpClient(auth_key="token", base_url="https://api.pipefy.com/graphql"))
>>> callable(service.updateAssigneeIds)
True
Return type:

Dict[str, Any]

updateLabelIds(card_id, label_ids)[source]

Update the labels of a card.

This method executes a GraphQL mutation to assign one or more labels to a card.

Parameters:

  • card_id (str) – str = Card ID to be updated

  • label_ids (List[str]) – list[str] = List of label IDs to assign to the card

Returns:

dict = Raw API response containing updated card data

Raises:
Example:
>>> service = CardService(PipefyHttpClient(auth_key="token", base_url="https://api.pipefy.com/graphql"))
>>> callable(service.updateLabelIds)
True
Return type:

Dict[str, Any]

getCardModel(card_id, query_body=None)[source]

Retrieve a card as a strongly-typed model.

This method retrieves structured card data and converts it into a Card instance.

Parameters:

  • card_id (str) – str = Card identifier

  • query_body (str | None) – str | None = Optional GraphQL fields

Returns:

Card = Parsed Card model instance

Raises:
Example:
>>> service = CardService(PipefyHttpClient(auth_key="token", base_url="https://api.pipefy.com/graphql"))
>>> callable(service.getCardModel)
True
Return type:

Card

listAllCardsPaginated(pipe_id, attributes=None)[source]

Retrieve all cards from a pipe as typed Card models using pagination.

This method performs cursor-based pagination and converts each card node into a Card model instance.

Parameters:

  • pipe_id (str) – str = Pipe ID from which cards will be retrieved

  • attributes (List[str] | None) – list[str] | None = List of attributes to include in the query. Defaults to [“id”, “title”]

Returns:

list[Card] = List of Card model instances

Raises:
Example:
>>> service = CardService(PipefyHttpClient(auth_key="token", base_url="https://api.pipefy.com/graphql"))
>>> callable(service.listAllCardsPaginated)
True
Return type:

List[Card]

listCardsFromPhasePaginated(phase_id, limit=50)[source]

Retrieve all cards from a specific phase as typed Card models using pagination.

This method performs cursor-based pagination within a phase and converts each card node into a Card model instance.

Parameters:

  • phase_id (str) – str = Phase ID from which cards will be retrieved

  • limit (int) – int = Number of items per page (page size)

Returns:

list[Card] = List of Card model instances

Raises:
Example:
>>> service = CardService(PipefyHttpClient(auth_key="token", base_url="https://api.pipefy.com/graphql"))
>>> callable(service.listCardsFromPhasePaginated)
True
Return type:

List[Card]

class pipebridge.FileService(client, card_service)[source]

Bases: object

Application service responsible for file operations in Pipefy.

This class acts as the orchestration layer between external interfaces (e.g., Facades) and internal domain flows.

ARCHITECTURAL ROLE:

  • Receives structured request objects (DTO pattern)

  • Delegates execution to domain flows

  • Manages dependencies and execution context

DESIGN PRINCIPLES:

  • No business logic

  • No parameter explosion

  • No data transformation logic

  • Strict delegation to flows

API DESIGN:
This service enforces a request-based API:

  • FileUploadRequest

  • FileDownloadRequest

This ensures:

  • Consistency across operations

  • Extensibility without breaking signatures

  • Strong typing and validation

Parameters:

  • client (PipefyHttpClient) – PipefyHttpClient = HTTP client instance

  • card_service (CardService) – CardService = Card service instance

Raises:

ValidationError – When dependencies are invalid

Example:
>>> from pipebridge.client.httpClient import PipefyHttpClient
>>> from pipebridge.service.card.cardService import CardService
>>> client = PipefyHttpClient("token", "https://api.pipefy.com/graphql")
>>> service = FileService(client, CardService(client))
>>> isinstance(service, FileService)
True
uploadFile(request, extra_rules=None, config=None, extra_steps_before=None, extra_steps_after=None)[source]

Uploads a file and attaches it to a Pipefy card field.

This method delegates execution to FileUploadFlow, which implements a pipeline-based upload mechanism.

Parameters:

  • request (FileUploadRequest) – FileUploadRequest = Upload request object

  • extra_rules (list[BaseRule] | None) – Optional[list[BaseRule]] = Set of custom rules constructed by the user. The rules will be applied before the upload flow.

  • config (UploadConfig | None) – Optional[UploadConfig] = Configuration for upload process

  • extra_steps_before (Sequence[BaseStep] | None) – Optional[Sequence[BaseStep]] = Additional custom steps executed before the built-in upload pipeline

  • extra_steps_after (Sequence[BaseStep] | None) – Optional[Sequence[BaseStep]] = Additional custom steps executed after the built-in upload pipeline

Returns:

FileUploadResult = Upload result metadata

Raises:

  • ValidationError – When request is invalid

  • FileFlowError – When the upload flow fails in a domain-specific way

  • WorkflowError – When rule execution or step execution fails at workflow level

  • RequestError – When a lower-level transport or API failure bubbles up

Example:
>>> callable(FileService.uploadFile)
True
Return type:

FileUploadResult

downloadAllAttachments(request)[source]

Downloads all attachments from a Pipefy card field.

This method retrieves file URLs from the specified card field, downloads each file, and saves them locally.

Parameters:

request (FileDownloadRequest) – FileDownloadRequest = Download request object

Returns:

list[Path] = List of saved file paths

Raises:
Example:
>>> callable(FileService.downloadAllAttachments)
True
Return type:

List[Path]

class pipebridge.PipeService(client)[source]

Bases: object

Service responsible for Pipe operations.

This service follows a layered architecture:

RAW → STRUCTURED → MODEL

It provides a unified and flexible interface to retrieve pipe data, including phases, labels, and users.

Parameters:

client (PipefyHttpClient) – PipefyHttpClient = HTTP client instance

getPipeRaw(pipe_id, query_body)[source]

Retrieve raw pipe data using a custom GraphQL query.

Parameters:

  • pipe_id (str) – str = Pipe identifier

  • query_body (str) – str = GraphQL fields to retrieve

Returns:

dict = Raw API response

Raises:
Example:
>>> callable(PipeService.getPipeRaw)
True
Return type:

Dict[str, Any]

getPipeFieldCatalog(pipe_id)[source]

Retrieve all configured phases and fields for a pipe as a model.

This method is intended for field discovery and schema inspection. It returns the canonical Pipe model, with phase fields and start form fields populated from a single pipe schema catalog query.

Parameters:

pipe_id (str) – str = Pipe identifier

Returns:

Pipe = Pipe model with phase fields populated

Raises:
Example:
>>> callable(PipeService.getPipeFieldCatalog)
True
Return type:

Pipe

getPipeStructured(pipe_id, query_body=None)[source]

Retrieve structured pipe data.

This method abstracts GraphQL complexity and returns a cleaned, consistent dictionary representing a pipe.

If no query_body is provided, a default query including phases, labels, users, and card count is used.

Parameters:

  • pipe_id (str) – str = Pipe identifier

  • query_body (str | None) – str | None = Optional GraphQL fields

Returns:

dict = Structured pipe data

Raises:
Example:
>>> callable(PipeService.getPipeStructured)
True
Return type:

Dict[str, Any]

getPipeModel(pipe_id, query_body=None)[source]

Retrieve a Pipe as a strongly-typed model.

This method converts structured pipe data into a Pipe object, enabling direct access to phases, labels, and users without loops.

Parameters:

  • pipe_id (str) – str = Pipe identifier

  • query_body (str | None) – str | None = Optional GraphQL fields

Returns:

Pipe = Parsed Pipe model

Raises:
Example:
>>> callable(PipeService.getPipeModel)
True
Return type:

Pipe

class pipebridge.PhaseService(client)[source]

Bases: object

Service responsible for handling Phase operations.

This service provides methods to retrieve raw, structured, and model-based representations of Pipefy Phases.

It follows a layered architecture:

RAW → STRUCTURED → MODEL

Parameters:

client (PipefyHttpClient) – PipefyHttpClient = HTTP client instance

getPhaseRaw(phase_id, query_body)[source]

Retrieve raw phase data using a custom GraphQL query body.

This method provides full flexibility for defining which fields should be retrieved from the Pipefy API.

Parameters:

  • phase_id (str) – str = Phase identifier

  • query_body (str) – str = GraphQL fields to retrieve

Returns:

dict = Raw API response

Raises:
Example:
>>> callable(PhaseService.getPhaseRaw)
True
Return type:

Dict[str, Any]

getPhaseStructured(phase_id, query_body=None)[source]

Retrieve structured phase data.

This method abstracts GraphQL complexity and returns a cleaned, consistent dictionary containing the requested phase data.

If no query_body is provided, a default structure including fields metadata is used.

Parameters:

  • phase_id (str) – str = Phase identifier

  • query_body (str | None) – str | None = Optional GraphQL fields

Returns:

dict = Structured phase data

Raises:
Example:
>>> callable(PhaseService.getPhaseStructured)
True
Return type:

Dict[str, Any]

getPhaseModel(phase_id, query_body=None)[source]

Retrieve a Phase as a strongly-typed model.

This method converts structured phase data into a Phase object, enabling advanced access patterns such as O(1) field lookup.

Parameters:

  • phase_id (str) – str = Phase identifier

  • query_body (str | None) – str | None = Optional GraphQL fields

Returns:

Phase = Parsed Phase model

Raises:
Example:
>>> callable(PhaseService.getPhaseModel)
True
Return type:

Phase

getPhaseField(phase_id, field_id)[source]

Retrieve phase schema metadata for a specific field.

This helper is phase-schema oriented. It does not inspect card payloads and does not depend on whether a field currently has a value in a card.

Parameters:

  • phase_id (str) – str = Phase identifier

  • field_id (str) – str = Field identifier

Returns:

PhaseField | None = Matching phase field when found

Return type:

PhaseField | None

hasPhaseField(phase_id, field_id)[source]

Check whether a field exists in a specific phase schema.

Parameters:

  • phase_id (str) – str = Phase identifier

  • field_id (str) – str = Field identifier

Returns:

bool = Whether the field exists in the phase schema

Return type:

bool

requirePhaseField(phase_id, field_id)[source]

Retrieve a phase field and fail semantically when it is missing.

Parameters:

  • phase_id (str) – str = Phase identifier

  • field_id (str) – str = Field identifier

Returns:

PhaseField = Requested phase field

Raises:

RequestError – When the field does not exist in the phase schema

Return type:

PhaseField

class pipebridge.ConnectorService(client, pipe_service=None, card_service=None)[source]

Bases: object

Service responsible for semantic connector discovery.

This service exposes connector-aware schema inspection and dynamic option discovery without requiring consumers to write raw GraphQL.

Connector option discovery is contextual. PipeBridge follows the same cards(… throughConnectors …) pattern used by the Pipefy UI and resolves the repo identifier dynamically from connector schema metadata.

Parameters:
listFields(pipe_id)[source]

Retrieve all connector fields configured in a pipe.

Parameters:

pipe_id (str) – str = Pipe identifier

Returns:

list[ConnectorFieldSpec] = Connector fields with origin metadata

Return type:

List[ConnectorFieldSpec]

getField(pipe_id, field_id, phase_id=None)[source]

Retrieve a connector field specification from a pipe.

Parameters:

  • pipe_id (str) – str = Pipe identifier

  • field_id (str) – str = Connector field identifier

  • phase_id (str | None) – str | None = Optional parent phase identifier

Returns:

ConnectorFieldSpec | None = Matching connector field when found

Return type:

ConnectorFieldSpec | None

requireField(pipe_id, field_id, phase_id=None)[source]

Retrieve a connector field specification and fail when it is missing.

Parameters:

  • pipe_id (str) – str = Pipe identifier

  • field_id (str) – str = Connector field identifier

  • phase_id (str | None) – str | None = Optional parent phase identifier

Returns:

ConnectorFieldSpec = Requested connector field

Return type:

ConnectorFieldSpec

listOptions(pipe_id, field_id, phase_id=None, search=None, limit=20)[source]

List dynamic options exposed by a connector field.

Parameters:

  • pipe_id (str) – str = Pipe identifier

  • field_id (str) – str = Connector field identifier

  • phase_id (str | None) – str | None = Optional phase identifier

  • search (str | None) – str | None = Optional title filter

  • limit (int) – int = Maximum number of options to return

Returns:

list[ConnectorOption] = Connector options

Return type:

List[ConnectorOption]

resolveOption(pipe_id, field_id, title, phase_id=None)[source]

Resolve a connector option by exact title.

Resolution is case-insensitive but exact after trimming. If zero matches are found, or more than one exact match exists, a validation error is raised.

Parameters:

  • pipe_id (str) – str = Pipe identifier

  • field_id (str) – str = Connector field identifier

  • title (str) – str = Desired connected item title

  • phase_id (str | None) – str | None = Optional phase identifier

Returns:

ConnectorOption = Unique resolved connector option

Return type:

ConnectorOption

validateConnectorSelection(field, value)[source]

Validate a connector payload against the connector field schema.

First-phase safe creation semantics require connector values to be passed as a list of connected item ids.

Validation performed:

  • field must be a connector

  • payload must be a list of non-empty ids

  • existing-item connection must be allowed

  • multiplicity must be respected

  • each id must exist in the connected repo referenced by the field

Parameters:

  • field (PhaseField) – PhaseField = Connector field schema

  • value (Any) – Any = Proposed connector payload

Returns:

list[str] = Normalized connected item ids

Return type:

List[str]

getCardValue(card_id, field_id)[source]

Retrieve a structured connector value for a card field.

This helper is schema-aware. If the connector exists in the card pipe but is absent from card.fields, an empty ConnectorValue is returned instead of treating the field as missing.

Parameters:

  • card_id (str) – str = Card identifier

  • field_id (str) – str = Connector field identifier

Returns:

ConnectorValue = Structured connector value

Return type:

ConnectorValue

setCardValue(card_id, field_id, item_ids)[source]

Replace the connected ids of a connector field.

Parameters:

  • card_id (str) – str = Card identifier

  • field_id (str) – str = Connector field identifier

  • item_ids (List[str]) – list[str] = Replacement connected item ids

Returns:

dict = Raw update response keyed by updated field

Return type:

Dict[str, Any]

addCardValue(card_id, field_id, item_ids)[source]

Add connected ids to a connector field using read-modify-write.

Parameters:

  • card_id (str) – str = Card identifier

  • field_id (str) – str = Connector field identifier

  • item_ids (List[str]) – list[str] = Connected item ids to add

Returns:

dict = Raw update response keyed by updated field

Return type:

Dict[str, Any]

removeCardValue(card_id, field_id, item_ids)[source]

Remove connected ids from a connector field using read-modify-write.

Parameters:

  • card_id (str) – str = Card identifier

  • field_id (str) – str = Connector field identifier

  • item_ids (List[str]) – list[str] = Connected item ids to remove

Returns:

dict = Raw update response keyed by updated field

Return type:

Dict[str, Any]

class pipebridge.FileUploadRequest(file_name, file_bytes, card_id, field_id, organization_id, replace_files=False, expected_phase_id=None)[source]

Bases: object

Represents a file upload request.

This class encapsulates all input data required to execute a file upload operation in Pipefy.

It is designed to:

  • Eliminate parameter explosion

  • Improve API readability

  • Provide validation at construction time

  • Enable extensibility without breaking method signatures

This object is immutable by design convention and should be treated as a command object.

Parameters:

  • file_name (str) – str = Name of the file to upload

  • file_bytes (bytes) – bytes = Binary content of the file

  • card_id (str) – str = Pipefy card identifier

  • field_id (str) – str = Target field identifier

  • organization_id (str) – str = Pipefy organization identifier

  • replace_files (bool) – bool = Whether to replace or append existing files

  • expected_phase_id (str | None) – str | None = Optional current-phase guard. When provided, the upload flow validates that the card is in this phase. When omitted or set to None, no current-phase validation is executed.

Raises:

ValidationError – When any parameter is invalid

Example:
>>> request = FileUploadRequest(
...     file_name="file.txt",
...     file_bytes=b"data",
...     card_id="123",
...     field_id="field_1",
...     organization_id="org_1"
... )
>>> isinstance(request, FileUploadRequest)
True
classmethod fromFile(file_path, card_id, field_id, organization_id, replace_files=False)[source]

Creates a FileUploadRequest from a file path.

Parameters:

  • file_path (str) – str = Local file path

  • card_id (str) – str

  • field_id (str) – str

  • organization_id (str) – str

  • replace_files (bool) – bool

Returns:

FileUploadRequest

Raises:
Example:
>>> import os, tempfile
>>> tmp = tempfile.NamedTemporaryFile(delete=False)
>>> _ = tmp.write(b"data")
>>> tmp.close()
>>> req = FileUploadRequest.fromFile(tmp.name, "1", "f", "org")
>>> isinstance(req, FileUploadRequest)
True
Return type:

FileUploadRequest

class pipebridge.FileDownloadRequest(card_id, field_id, output_dir)[source]

Bases: object

Represents a request to download attachments from a Pipefy card.

This object encapsulates all required parameters for downloading files from a card field.

DESIGN PURPOSE:

  • Replace primitive parameter passing

  • Provide validation at construction time

  • Improve API consistency with upload flow

Parameters:

  • card_id (str) – str = Card identifier

  • field_id (str) – str = Field identifier

  • output_dir (str) – str = Output directory path

Raises:

ValidationError – When parameters are invalid

Example:
>>> req = FileDownloadRequest("1", "field", "/tmp")
>>> isinstance(req, FileDownloadRequest)
True
class pipebridge.UploadConfig(retry=None, circuit=None, validate_field_in_current_phase=True)[source]

Bases: object

Configuration for upload flow execution policies.

This object preserves the existing public API while internally supporting the refactored workflow/policy engine.

Parameters:

  • retry (RetryConfig | None) – Optional[RetryConfig] = Retry configuration

  • circuit (CircuitBreakerConfig | None) – Optional[CircuitBreakerConfig] = Circuit breaker configuration

  • validate_field_in_current_phase (bool) – bool = Whether upload field existence must be validated strictly against the card current phase schema. When disabled, the field may exist anywhere in the pipe schema, including start form or another phase.

Example:
>>> cfg = UploadConfig()
>>> isinstance(cfg.retry, RetryConfig)
True
class pipebridge.CardUpdateConfig(retry=None, circuit=None, validate_phase=True, validate_field_existence=True, validate_field_options=True, validate_field_type=True, validate_field_format=False, load_phase_schema=False)[source]

Bases: object

Configuration for card update flows.

Parameters:

  • retry (RetryConfig | None)

  • circuit (CircuitBreakerConfig | None)

  • validate_phase (bool)

  • validate_field_existence (bool)

  • validate_field_options (bool)

  • validate_field_type (bool)

  • validate_field_format (bool)

  • load_phase_schema (bool)

requiresPhaseSchema()[source]

Determine whether the update flow needs the current phase schema.

Returns:

bool = Whether the flow must load phase metadata

Return type:

bool

class pipebridge.CardMoveConfig(retry=None, circuit=None, validate_required_fields=True, load_target_phase_schema=True)[source]

Bases: object

Configuration for safe card phase transitions.

Parameters:

  • retry (RetryConfig | None) – RetryConfig | None = Retry policy configuration

  • circuit (CircuitBreakerConfig | None) – CircuitBreakerConfig | None = Circuit breaker configuration

  • validate_required_fields (bool) – bool = Whether the flow must validate required fields from the destination phase before moving the card

  • load_target_phase_schema (bool) – bool = Whether to load the destination phase schema even when required-field validation is disabled

requiresTargetPhaseSchema()[source]

Determine whether destination phase metadata must be loaded.

Returns:

bool = Whether the move flow needs the target phase schema

Example:
>>> CardMoveConfig().requiresTargetPhaseSchema()
True
Return type:

bool

Facade

class pipebridge.facade.pipefyFacade.CardsFacade(service)[source]

Bases: object

Public facade for card operations exposed by the SDK.

This facade keeps the public API stable while delegating business logic to pipebridge.service.card.cardService.CardService.

Parameters:

service (CardService)

get(card_id)[source]

Retrieve a card as a model.

Important:

card.fields reflects the fields payload returned by Pipefy for the card query. In practice this should be treated as the set of materialized field values exposed by the API for that card, not as the complete schema of the current phase or the pipe.

Use phase or pipe schema helpers when you need to validate field existence independently from the current card payload.

Parameters:

card_id (str) – str = Card identifier

Returns:

Card = Card model instance

Raises:
Example:
>>> callable(CardsFacade.get)
True
Return type:

Card

create(pipe_id, title, fields=None)[source]

Create a new card.

Parameters:

  • pipe_id (str) – str = Pipe identifier

  • title (str) – str = Card title

  • fields (Dict[str, Any] | None) – dict[str, Any] | None = Field values

Returns:

dict = Raw API response

Raises:
Example:
>>> callable(CardsFacade.create)
True
Return type:

Dict[str, Any]

createSafely(pipe_id, title, fields=None)[source]

Create a new card after validating the payload against start form schema.

Parameters:

  • pipe_id (str) – str = Pipe identifier

  • title (str) – str = Card title

  • fields (Dict[str, Any] | None) – dict[str, Any] | None = Start form field values

Returns:

dict = Raw API response

Raises:
Return type:

Dict[str, Any]

move(card_id, phase_id)[source]

Move a card to another phase.

Parameters:

  • card_id (str) – str = Card identifier

  • phase_id (str) – str = Destination phase ID

Returns:

dict = Raw API response

Raises:
Example:
>>> callable(CardsFacade.move)
True
Return type:

Dict[str, Any]

moveSafely(card_id, destination_phase_id, expected_current_phase_id=None, config=None, extra_rules=None)[source]

Move a card to another phase after validating transition readiness.

This is the recommended public entry point when the caller wants the SDK to validate destination required fields before moving the card.

Parameters:

  • card_id (str) – str = Card identifier

  • destination_phase_id (str) – str = Destination phase identifier

  • expected_current_phase_id (str | None) – str | None = Optional expected source phase identifier

  • config (CardMoveConfig | None) – CardMoveConfig | None = Optional move flow configuration

  • extra_rules (List[BaseRule] | None) – list[BaseRule] | None = Optional extra rules

Returns:

dict = Raw move API response

Return type:

Dict[str, Any]

delete(card_id)[source]

Delete a card.

Parameters:

card_id (str) – str = Card identifier

Returns:

dict = Raw API response

Raises:
Example:
>>> callable(CardsFacade.delete)
True
Return type:

Dict[str, Any]

list(pipe_id)[source]

List all cards from a pipe.

Parameters:

pipe_id (str) – str = Pipe identifier

Returns:

list[Card] = List of card models

Raises:
Example:
>>> callable(CardsFacade.list)
True
Return type:

List[Card]

search(pipe_id, title)[source]

Search cards by title.

Parameters:

  • pipe_id (str) – str = Pipe identifier

  • title (str) – str = Search term

Returns:

list[dict] = Matching cards

Raises:
Example:
>>> callable(CardsFacade.search)
True
Return type:

List[Dict[str, Any]]

updateField(card_id, field_id, value, config=None, extra_rules=None, extra_handlers=None)[source]

Update a single card field through the public facade.

Parameters:

  • card_id (str) – str = Card identifier

  • field_id (str) – str = Logical field identifier

  • value (Any) – Any = New field value

  • config (CardUpdateConfig | None) – CardUpdateConfig | None = Optional update flow configuration

  • extra_rules (List[BaseRule] | None) – list[BaseRule] | None = Optional custom validation rules

  • extra_handlers (Dict[str, BaseCardFieldUpdateHandler] | None)

Returns:

dict = Raw API response keyed by updated field

Raises:
Return type:

Dict[str, Any]

updateFields(card_id, fields, expected_phase_id=None, config=None, extra_rules=None, extra_handlers=None)[source]

Update multiple card fields through the public facade.

Parameters:

  • card_id (str) – str = Card identifier

  • fields (Dict[str, Any]) – dict[str, Any] = Field values mapped by field ID

  • expected_phase_id (str | None) – str | None = Optional expected phase ID

  • config (CardUpdateConfig | None) – CardUpdateConfig | None = Optional update flow configuration

  • extra_rules (List[BaseRule] | None) – list[BaseRule] | None = Optional custom validation rules

  • extra_handlers (Dict[str, BaseCardFieldUpdateHandler] | None)

Returns:

dict = Raw API responses keyed by updated field

Raises:
Return type:

Dict[str, Any]

invalidateSchemaCache(pipe_id=None)[source]

Invalidate the in-memory pipe schema cache used by card update flows.

Parameters:

pipe_id (str | None) – str | None = Specific pipe identifier. When omitted, the whole schema cache is invalidated.

Returns:

None

Return type:

None

getSchemaCacheStats()[source]

Retrieve aggregate statistics about the card schema cache.

Returns:

dict = Cache statistics

Return type:

Dict[str, Any]

getSchemaCacheEntryInfo(pipe_id)[source]

Retrieve metadata about a specific cached schema entry.

Parameters:

pipe_id (str) – str = Pipe identifier

Returns:

dict | None = Cache entry metadata

Return type:

Dict[str, Any] | None

class pipebridge.facade.pipefyFacade.CardsRawFacade(service)[source]

Bases: object

Low-level card operations (raw GraphQL access).

Parameters:

service (CardService) – CardService = Card service instance

Example:
>>> callable(CardsRawFacade.get)
True
get(card_id, query)[source]

Retrieve raw card data using custom GraphQL query.

Parameters:

  • card_id (str) – str = Card identifier

  • query (str) – str = GraphQL query

Returns:

dict = Raw API response

Raises:
Example:
>>> callable(CardsRawFacade.get)
True
Return type:

Dict[str, Any]

class pipebridge.facade.pipefyFacade.CardsStructuredFacade(service)[source]

Bases: object

Intermediate-level card operations (structured response).

Parameters:

service (CardService) – CardService = Card service instance

Example:
>>> callable(CardsStructuredFacade.get)
True
get(card_id, query_body=None)[source]

Retrieve structured card data.

Parameters:

  • card_id (str) – str = Card identifier

  • query_body (str | None) – str | None = Custom query body

Returns:

dict = Structured response

Raises:
Example:
>>> callable(CardsStructuredFacade.get)
True
Return type:

Dict[str, Any]

class pipebridge.facade.pipefyFacade.PhasesFacade(service, card_service)[source]

Bases: object

Public facade for phase retrieval operations.

This facade exposes high-level access to phase data while keeping the underlying service layer internal to the SDK.

Parameters:
get(phase_id, query_body=None)[source]

Retrieve a phase model.

Parameters:

  • phase_id (str) – str = Phase identifier

  • query_body (str | None) – str | None = Optional GraphQL query body

Returns:

Phase = Phase model instance

Raises:
Example:
>>> callable(PhasesFacade.get)
True
Return type:

Phase

getField(phase_id, field_id)[source]

Retrieve schema metadata for a field inside a specific phase.

This helper is phase-schema oriented. It should be used when the caller needs to know whether a field belongs to a phase regardless of whether a card currently exposes a value for that field.

Parameters:

  • phase_id (str) – str = Phase identifier

  • field_id (str) – str = Field identifier

Returns:

PhaseField | None = Matching phase field when found

Return type:

PhaseField | None

hasField(phase_id, field_id)[source]

Check whether a field exists in a specific phase schema.

Parameters:

  • phase_id (str) – str = Phase identifier

  • field_id (str) – str = Field identifier

Returns:

bool = Whether the field exists in the phase schema

Return type:

bool

requireField(phase_id, field_id)[source]

Retrieve a phase field and fail semantically when it does not exist.

Parameters:

  • phase_id (str) – str = Phase identifier

  • field_id (str) – str = Field identifier

Returns:

PhaseField = Requested phase field

Return type:

PhaseField

listCards(phase_id)[source]

List all cards from a phase.

Parameters:

phase_id (str) – str = Phase identifier

Returns:

list[Card] = Cards in phase

Raises:

Exception – Propagates service exceptions

Example:
>>> callable(PhasesFacade.listCards)
True
Return type:

List[Card]

class pipebridge.facade.pipefyFacade.PhasesRawFacade(service)[source]

Bases: object

Low-level phase operations (raw GraphQL access).

Parameters:

service (PhaseService) – PhaseService = Phase service instance

Example:
>>> callable(PhasesRawFacade.get)
True
get(phase_id, query)[source]

Retrieve raw phase data.

Parameters:

  • phase_id (str) – str = Phase identifier

  • query (str) – str = GraphQL query body

Returns:

dict = Raw API response

Raises:
Example:
>>> callable(PhasesRawFacade.get)
True
Return type:

Dict[str, Any]

class pipebridge.facade.pipefyFacade.PhasesStructuredFacade(service)[source]

Bases: object

Intermediate-level phase operations (structured response).

Parameters:

service (PhaseService) – PhaseService = Phase service instance

Example:
>>> callable(PhasesStructuredFacade.get)
True
get(phase_id, query_body=None)[source]

Retrieve structured phase data.

Parameters:

  • phase_id (str) – str = Phase identifier

  • query_body (str | None) – str | None = Custom query body

Returns:

dict = Structured response

Raises:
Example:
>>> callable(PhasesStructuredFacade.get)
True
Return type:

Dict[str, Any]

class pipebridge.facade.pipefyFacade.PipesFacade(service)[source]

Bases: object

Public facade for pipe retrieval and schema discovery operations.

This facade is the recommended entry point for consumers that need pipe models, raw responses, or field catalog discovery.

Parameters:

service (PipeService)

get(pipe_id, query_body=None)[source]

Retrieve a pipe model.

Parameters:

  • pipe_id (str) – str = Pipe identifier

  • query_body (str | None) – str | None = Optional GraphQL query body

Returns:

Pipe = PipeService model instance

Raises:
Example:
>>> callable(PipesFacade.get)
True
Return type:

Pipe

getFieldCatalog(pipe_id)[source]

Retrieve all phases and configured fields for a pipe.

This is the discovery-oriented variant of get(), useful for building update coverage and inspecting the schema available in a pipe.

The returned catalog includes:

  • start form fields

  • phase fields

  • connector schema metadata when exposed by Pipefy

Parameters:

pipe_id (str) – str = Pipe identifier

Returns:

Pipe = Pipe model with phase fields populated

Raises:
Example:
>>> callable(PipesFacade.getFieldCatalog)
True
Return type:

Pipe

class pipebridge.facade.pipefyFacade.PipesRawFacade(service)[source]

Bases: object

Low-level pipe operations (raw GraphQL access).

Parameters:

service (PipeService) – PipeService = Pipe service instance

Example:
>>> callable(PipesRawFacade.get)
True
get(pipe_id, query)[source]

Retrieve raw pipe data.

Parameters:

  • pipe_id (str) – str = Pipe identifier

  • query (str) – str = GraphQL query body

Returns:

dict = Raw API response

Raises:
Example:
>>> callable(PipesRawFacade.get)
True
Return type:

Dict[str, Any]

class pipebridge.facade.pipefyFacade.PipesStructuredFacade(service)[source]

Bases: object

Intermediate-level pipe operations (structured response).

Parameters:

service (PipeService) – PipeService = Pipe service instance

Example:
>>> callable(PipesStructuredFacade.get)
True
get(pipe_id, query_body=None)[source]

Retrieve structured pipe data.

Parameters:

  • pipe_id (str) – str = Pipe identifier

  • query_body (str | None) – str | None = Optional GraphQL fields

Returns:

dict = Structured response

Raises:
Example:
>>> callable(PipesStructuredFacade.get)
True
Return type:

Dict[str, Any]

class pipebridge.facade.pipefyFacade.ConnectorsFacade(service)[source]

Bases: object

Public facade for semantic connector discovery operations.

This facade exposes connector schema inspection and dynamic option discovery without requiring consumers to write raw GraphQL.

Parameters:

service (ConnectorService)

listFields(pipe_id)[source]

List connector fields configured in a pipe.

Parameters:

pipe_id (str) – str = Pipe identifier

Returns:

list[ConnectorFieldSpec] = Connector fields with origin metadata

Return type:

List[ConnectorFieldSpec]

getField(pipe_id, field_id, phase_id=None)[source]

Retrieve a connector field specification from a pipe.

Parameters:

  • pipe_id (str) – str = Pipe identifier

  • field_id (str) – str = Connector field identifier

  • phase_id (str | None) – str | None = Optional parent phase identifier

Returns:

ConnectorFieldSpec | None = Matching connector field when found

Return type:

ConnectorFieldSpec | None

requireField(pipe_id, field_id, phase_id=None)[source]

Retrieve a connector field specification and fail when it is missing.

Parameters:

  • pipe_id (str) – str = Pipe identifier

  • field_id (str) – str = Connector field identifier

  • phase_id (str | None) – str | None = Optional parent phase identifier

Returns:

ConnectorFieldSpec = Requested connector field

Return type:

ConnectorFieldSpec

listOptions(pipe_id, field_id, phase_id=None, search=None, limit=20)[source]

List connector options from the connected repo.

Parameters:

  • pipe_id (str) – str = Pipe identifier

  • field_id (str) – str = Connector field identifier

  • phase_id (str | None) – str | None = Optional phase identifier

  • search (str | None) – str | None = Optional title filter

  • limit (int) – int = Maximum number of results

Returns:

list[ConnectorOption] = Connector options

Return type:

List[ConnectorOption]

resolveOption(pipe_id, field_id, title, phase_id=None)[source]

Resolve a connector option by exact title.

Parameters:

  • pipe_id (str) – str = Pipe identifier

  • field_id (str) – str = Connector field identifier

  • title (str) – str = Desired option title

  • phase_id (str | None) – str | None = Optional phase identifier

Returns:

ConnectorOption = Unique resolved connector option

Return type:

ConnectorOption

getCardValue(card_id, field_id)[source]

Retrieve the structured connector value of a card field.

Parameters:

  • card_id (str) – str = Card identifier

  • field_id (str) – str = Connector field identifier

Returns:

ConnectorValue = Structured connector value

Return type:

ConnectorValue

setCardValue(card_id, field_id, item_ids)[source]

Replace the connected ids of a connector field.

Parameters:

  • card_id (str) – str = Card identifier

  • field_id (str) – str = Connector field identifier

  • item_ids (List[str]) – list[str] = Replacement connected item ids

Returns:

dict = Raw update response keyed by updated field

Return type:

Dict[str, Any]

addCardValue(card_id, field_id, item_ids)[source]

Add connected ids to a connector field.

Parameters:

  • card_id (str) – str = Card identifier

  • field_id (str) – str = Connector field identifier

  • item_ids (List[str]) – list[str] = Connected item ids to add

Returns:

dict = Raw update response keyed by updated field

Return type:

Dict[str, Any]

removeCardValue(card_id, field_id, item_ids)[source]

Remove connected ids from a connector field.

Parameters:

  • card_id (str) – str = Card identifier

  • field_id (str) – str = Connector field identifier

  • item_ids (List[str]) – list[str] = Connected item ids to remove

Returns:

dict = Raw update response keyed by updated field

Return type:

Dict[str, Any]

class pipebridge.facade.pipefyFacade.FilesFacade(service)[source]

Bases: object

Facade for file operations in Pipefy.

This class provides a unified and simplified interface for file upload and download operations.

DESIGN PRINCIPLES:

  • Uses request objects (no primitive parameter explosion)

  • Delegates all logic to FileService

  • Maintains a stable API boundary

Parameters:

service (FileService) – FileService

Example:
>>> callable(FilesFacade.uploadFile)
True
uploadFile(request, extra_rules=None, config=None, extra_steps_before=None, extra_steps_after=None)[source]

Uploads a file using a structured request object.

Parameters:

  • request (FileUploadRequest) – FileUploadRequest

  • extra_rules (list[BaseRule] | None) – Optional[list[BaseRule]] = Set of custom rules constructed by te user. The rules will be applied before the upload flow.

  • config (UploadConfig | None) – Optional[UploadConfig] = Configuration for upload process

  • extra_steps_before (Sequence[BaseStep] | None) – Optional[Sequence[BaseStep]] = Additional custom steps executed before the built-in upload pipeline

  • extra_steps_after (Sequence[BaseStep] | None) – Optional[Sequence[BaseStep]] = Additional custom steps executed after the built-in upload pipeline

Returns:

FileUploadResult

Raises:
Example:
>>> callable(FilesFacade.uploadFile)
True
Return type:

FileUploadResult

downloadAllAttachments(request)[source]

Downloads all attachments from a card field.

This method delegates execution to FileService using a structured request object.

Parameters:

request (FileDownloadRequest) – FileDownloadRequest

Returns:

list[Path] = List of saved file paths

Raises:
Example:
>>> callable(FilesFacade.downloadAllAttachments)
True
Return type:

List[Path]

class pipebridge.facade.pipefyFacade.PipeBridge(token, base_url, pipe_schema_cache_ttl_seconds=300, transport_config=None)[source]

Bases: object

Root public facade of the Pipefy SDK.

This class wires the HTTP client and all domain facades, exposing a cohesive and stable API surface for cards, phases, pipes, and files.

Parameters:

Card Service

class pipebridge.service.card.cardService.CardService(client, pipe_schema_cache_ttl_seconds=300)[source]

Bases: object

Service responsible for card operations in Pipefy.

This service provides a complete abstraction over Pipefy GraphQL for managing cards, including creation, retrieval, updates, and movement across phases.

param client:

PipefyHttpClient = HTTP client instance

example:
>>> service = CardService(PipefyHttpClient(auth_key="token", base_url="https://api.pipefy.com/graphql"))
>>> isinstance(service, CardService)
True
Parameters:
invalidatePipeSchemaCache(pipe_id=None)[source]

Invalidate the in-memory pipe schema cache.

Parameters:

pipe_id (str | None) – str | None = Specific pipe identifier. When omitted, the entire schema cache is invalidated.

Returns:

None

Return type:

None

getPipeSchemaCacheStats()[source]

Retrieve aggregate statistics from the pipe schema cache.

Returns:

dict = Cache statistics

Return type:

Dict[str, Any]

getPipeSchemaCacheEntryInfo(pipe_id)[source]

Retrieve metadata about a specific cached pipe schema entry.

Parameters:

pipe_id (str) – str = Pipe identifier

Returns:

dict | None = Cache entry metadata

Return type:

Dict[str, Any] | None

createCard(pipe_id, title, fields=None)[source]

Create a new card in a pipe.

This method creates a card in the specified pipe and optionally sets initial field values.

Parameters:

  • pipe_id (str) – str = Pipe identifier

  • title (str) – str = Card title

  • fields (Dict[str, Any] | None) – dict[str, Any] | None = Field values mapped by field_id

Returns:

dict = Raw API response containing created card metadata

Raises:
Example:
>>> service = CardService(PipefyHttpClient(auth_key="token", base_url="https://api.pipefy.com/graphql"))
>>> callable(service.createCard)
True
Return type:

Dict[str, Any]

createCardSafely(pipe_id, title, fields=None)[source]

Create a new card after validating input against the pipe start form schema.

Parameters:

  • pipe_id (str) – str = Pipe identifier

  • title (str) – str = Card title

  • fields (Dict[str, Any] | None) – dict[str, Any] | None = Start form field values

Returns:

dict = Raw API response containing created card metadata

Raises:
Return type:

Dict[str, Any]

getRawCard(card_id, query_body)[source]

Execute a raw GraphQL query for a card.

This method allows full control over the GraphQL query structure.

Parameters:

  • card_id (str) – str = Card identifier

  • query_body (str) – str = GraphQL body inside card query

Returns:

dict = Raw API response

Raises:
Example:
>>> service = CardService(PipefyHttpClient(auth_key="token", base_url="https://api.pipefy.com/graphql"))
>>> callable(service.getRawCard)
True
Return type:

Dict[str, Any]

getCardStructured(card_id, query_body=None)[source]

Retrieve structured card data.

This method abstracts GraphQL complexity by providing a default query that includes commonly used fields such as phases, fields, labels, assignees, and attachments.

Parameters:

  • card_id (str) – str = Card identifier

  • query_body (str | None) – str | None = Custom GraphQL body override

Returns:

dict = Structured card data

Raises:
Example:
>>> service = CardService(PipefyHttpClient(auth_key="token", base_url="https://api.pipefy.com/graphql"))
>>> callable(service.getCardStructured)
True
Return type:

Dict[str, Any]

getCardPipeField(card_id, field_id)[source]

Retrieve schema metadata for a card field anywhere in the parent pipe.

This helper is schema-oriented. It does not depend on whether the field currently appears in card.fields.

Parameters:

  • card_id (str) – str = Card identifier

  • field_id (str) – str = Field identifier

Returns:

PhaseField | None = Matching schema field when found

Return type:

PhaseField | None

getCardCurrentPhaseField(card_id, field_id)[source]

Retrieve schema metadata for a field in the card current phase.

This helper is schema-oriented. It does not depend on whether the field currently appears in card.fields.

Parameters:

  • card_id (str) – str = Card identifier

  • field_id (str) – str = Field identifier

Returns:

PhaseField | None = Matching current-phase field when found

Return type:

PhaseField | None

listCardAttachments(card_id)[source]

Retrieve all attachments exposed by the Pipefy card attachments surface.

This helper is more robust than reading attachment URLs from card.fields because it also works for attachment fields that are not materialized in the current card field payload.

Parameters:

card_id (str) – str = Card identifier

Returns:

list[dict[str, Any]] = Attachment payload items

Return type:

List[Dict[str, Any]]

listCardAttachmentsByField(card_id, field_id)[source]

Retrieve attachments associated with a specific field in a card.

Parameters:

  • card_id (str) – str = Card identifier

  • field_id (str) – str = Field identifier

Returns:

list[dict[str, Any]] = Attachment payload items for the field

Return type:

List[Dict[str, Any]]

moveCardToPhase(card_id, phase_id)[source]

Move a card to another phase in Pipefy.

This method executes a GraphQL mutation to move a card from its current phase to a destination phase.

Parameters:

  • card_id (str) – str = Unique identifier of the card to be moved

  • phase_id (str) – str = Destination phase ID

Returns:

dict = Raw API response containing updated card reference

Raises:
Example:
>>> service = CardService(PipefyHttpClient(auth_key="token", base_url="https://api.pipefy.com/graphql"))
>>> callable(service.moveCardToPhase)
True
Return type:

Dict[str, Any]

moveCardToPhaseSafely(card_id, destination_phase_id, expected_current_phase_id=None, config=None, extra_rules=None)[source]

Move a card to another phase after validating transition readiness.

This safe flow can validate destination required fields before executing the low-level move mutation.

Parameters:

  • card_id (str) – str = Card identifier

  • destination_phase_id (str) – str = Destination phase identifier

  • expected_current_phase_id (str | None) – str | None = Optional expected source phase identifier

  • config (CardMoveConfig | None) – CardMoveConfig | None = Optional flow configuration

  • extra_rules (List[BaseRule] | None) – list[BaseRule] | None = Optional additional rules

Returns:

dict = Raw API move response

Raises:
Return type:

Dict[str, Any]

searchCards(pipe_id, title)[source]

Search for cards in a pipe by title.

This method executes a GraphQL query to find cards whose titles match the provided search term.

Parameters:

  • pipe_id (str) – str = Pipe ID where the search will be performed

  • title (str) – str = Title or partial title used as search filter

Returns:

list[dict] = List of card nodes containing basic information (id, title)

Raises:
Example:
>>> service = CardService(PipefyHttpClient(auth_key="token", base_url="https://api.pipefy.com/graphql"))
>>> callable(service.searchCards)
True
Return type:

List[Dict[str, Any]]

relateCards(parent_id, child_id, source_id)[source]

Create a relationship between two cards in Pipefy.

This method executes a GraphQL mutation to establish a relationship between a parent card and a child card using a Pipe relation.

Parameters:

  • parent_id (str) – str = Parent card ID (source of the relationship)

  • child_id (str) – str = Child card ID (target of the relationship)

  • source_id (str) – str = Pipe relation ID used to define the relationship

Returns:

dict = Raw API response containing the created relation ID

Raises:
Example:
>>> service = CardService(PipefyHttpClient(auth_key="token", base_url="https://api.pipefy.com/graphql"))
>>> callable(service.relateCards)
True
Return type:

Dict[str, Any]

listAllCards(pipe_id, attributes=None, filters=None, only_phase_name=None)[source]

Retrieve all cards from a pipe using cursor-based pagination.

This method iteratively fetches all cards from a given pipe using GraphQL pagination. It supports optional filtering and phase-based restriction.

Parameters:

  • pipe_id (str) – str = Pipe ID from which cards will be retrieved

  • attributes (List[str] | None) – list[str] | None = List of attributes to include in the query. Defaults to [“id”, “title”]

  • filters (Dict[str, Any] | None) – dict | None = GraphQL filter object to refine results

  • only_phase_name (str | None) – str | None = If provided, only cards belonging to this phase name will be returned

Returns:

list[dict] = List of card nodes matching the query criteria

Raises:
Example:
>>> service = CardService(PipefyHttpClient(auth_key="token", base_url="https://api.pipefy.com/graphql"))
>>> callable(service.listAllCards)
True
Return type:

List[Dict[str, Any]]

listCardsFromPhase(phase_id, limit=50)[source]

Retrieve all cards from a specific phase using cursor-based pagination.

This method iteratively fetches cards from a given phase, handling pagination internally until all cards are retrieved.

Parameters:

  • phase_id (str) – str = Phase ID from which cards will be retrieved

  • limit (int) – int = Number of cards per request (page size)

Returns:

list[dict] = List of card nodes containing basic information (id, title)

Raises:
Example:
>>> service = CardService(PipefyHttpClient(auth_key="token", base_url="https://api.pipefy.com/graphql"))
>>> callable(service.listCardsFromPhase)
True
Return type:

List[Dict[str, Any]]

deleteCard(card_id)[source]

Delete a card from Pipefy.

This method executes a GraphQL mutation to permanently remove a card identified by its ID.

Parameters:

card_id (str) – str = Unique identifier of the card to be deleted

Returns:

dict = Raw API response indicating operation success

Raises:
Example:
>>> service = CardService(PipefyHttpClient(auth_key="token", base_url="https://api.pipefy.com/graphql"))
>>> callable(service.deleteCard)
True
Return type:

Dict[str, Any]

updateField(card_id, field_id, value, config=None, extra_rules=None, extra_handlers=None)[source]

Update a single field of a card.

This is a convenience wrapper over updateFields.

Parameters:

  • card_id (str) – str = Card identifier

  • field_id (str) – str = Field identifier

  • value (str) – str = Field value

  • config (CardUpdateConfig | None)

  • extra_rules (List[BaseRule] | None)

  • extra_handlers (Dict[str, BaseCardFieldUpdateHandler] | None)

Returns:

dict = Raw API response

Raises:
Example:
>>> service = CardService(PipefyHttpClient(auth_key="token", base_url="https://api.pipefy.com/graphql"))
>>> callable(service.updateField)
True
Return type:

Dict[str, Any]

updateFields(card_id, fields, expected_phase_id=None, config=None, extra_rules=None, extra_handlers=None)[source]

Updates multiple fields of a card using Pipefy API.

This method performs:

  • Phase validation (optional)

  • Field validation (required, options, existence)

  • Dynamic handling for attachments vs standard fields

  • Execution of GraphQL mutations per field

Parameters:

  • card_id (str) – str = Card identifier

  • fields (Dict[str, Any]) – dict[str, Any] = Mapping of field_id to new values

  • expected_phase_id (str | None) – Optional[str] = Expected phase ID for validation

  • config (CardUpdateConfig | None) – Optional[CardUpdateConfig] = Optional flow configuration

  • extra_rules (List[BaseRule] | None) – Optional[list[BaseRule]] = Optional custom validation rules

  • extra_handlers (Dict[str, BaseCardFieldUpdateHandler] | None)

Returns:

dict = Aggregated API responses per field

Raises:
Example:
>>> service = CardService(PipefyHttpClient(auth_key="token", base_url="https://api.pipefy.com/graphql"))
>>> callable(service._buildAttachmentMutation)
True
Return type:

Dict[str, Any]

updateAssigneeIds(card_id, assignee_ids)[source]

Update the assignees of a card.

This method executes a GraphQL mutation to assign one or more users to a card.

Parameters:

  • card_id (str) – str = Card ID to be updated

  • assignee_ids (List[str]) – list[str] = List of user IDs to assign to the card

Returns:

dict = Raw API response containing updated card data

Raises:
Example:
>>> service = CardService(PipefyHttpClient(auth_key="token", base_url="https://api.pipefy.com/graphql"))
>>> callable(service.updateAssigneeIds)
True
Return type:

Dict[str, Any]

updateLabelIds(card_id, label_ids)[source]

Update the labels of a card.

This method executes a GraphQL mutation to assign one or more labels to a card.

Parameters:

  • card_id (str) – str = Card ID to be updated

  • label_ids (List[str]) – list[str] = List of label IDs to assign to the card

Returns:

dict = Raw API response containing updated card data

Raises:
Example:
>>> service = CardService(PipefyHttpClient(auth_key="token", base_url="https://api.pipefy.com/graphql"))
>>> callable(service.updateLabelIds)
True
Return type:

Dict[str, Any]

getCardModel(card_id, query_body=None)[source]

Retrieve a card as a strongly-typed model.

This method retrieves structured card data and converts it into a Card instance.

Parameters:

  • card_id (str) – str = Card identifier

  • query_body (str | None) – str | None = Optional GraphQL fields

Returns:

Card = Parsed Card model instance

Raises:
Example:
>>> service = CardService(PipefyHttpClient(auth_key="token", base_url="https://api.pipefy.com/graphql"))
>>> callable(service.getCardModel)
True
Return type:

Card

listAllCardsPaginated(pipe_id, attributes=None)[source]

Retrieve all cards from a pipe as typed Card models using pagination.

This method performs cursor-based pagination and converts each card node into a Card model instance.

Parameters:

  • pipe_id (str) – str = Pipe ID from which cards will be retrieved

  • attributes (List[str] | None) – list[str] | None = List of attributes to include in the query. Defaults to [“id”, “title”]

Returns:

list[Card] = List of Card model instances

Raises:
Example:
>>> service = CardService(PipefyHttpClient(auth_key="token", base_url="https://api.pipefy.com/graphql"))
>>> callable(service.listAllCardsPaginated)
True
Return type:

List[Card]

listCardsFromPhasePaginated(phase_id, limit=50)[source]

Retrieve all cards from a specific phase as typed Card models using pagination.

This method performs cursor-based pagination within a phase and converts each card node into a Card model instance.

Parameters:

  • phase_id (str) – str = Phase ID from which cards will be retrieved

  • limit (int) – int = Number of items per page (page size)

Returns:

list[Card] = List of Card model instances

Raises:
Example:
>>> service = CardService(PipefyHttpClient(auth_key="token", base_url="https://api.pipefy.com/graphql"))
>>> callable(service.listCardsFromPhasePaginated)
True
Return type:

List[Card]

Phase Service

class pipebridge.service.phase.phaseService.PhaseService(client)[source]

Bases: object

Service responsible for handling Phase operations.

This service provides methods to retrieve raw, structured, and model-based representations of Pipefy Phases.

It follows a layered architecture:

RAW → STRUCTURED → MODEL

Parameters:

client (PipefyHttpClient) – PipefyHttpClient = HTTP client instance

getPhaseRaw(phase_id, query_body)[source]

Retrieve raw phase data using a custom GraphQL query body.

This method provides full flexibility for defining which fields should be retrieved from the Pipefy API.

Parameters:

  • phase_id (str) – str = Phase identifier

  • query_body (str) – str = GraphQL fields to retrieve

Returns:

dict = Raw API response

Raises:
Example:
>>> callable(PhaseService.getPhaseRaw)
True
Return type:

Dict[str, Any]

getPhaseStructured(phase_id, query_body=None)[source]

Retrieve structured phase data.

This method abstracts GraphQL complexity and returns a cleaned, consistent dictionary containing the requested phase data.

If no query_body is provided, a default structure including fields metadata is used.

Parameters:

  • phase_id (str) – str = Phase identifier

  • query_body (str | None) – str | None = Optional GraphQL fields

Returns:

dict = Structured phase data

Raises:
Example:
>>> callable(PhaseService.getPhaseStructured)
True
Return type:

Dict[str, Any]

getPhaseModel(phase_id, query_body=None)[source]

Retrieve a Phase as a strongly-typed model.

This method converts structured phase data into a Phase object, enabling advanced access patterns such as O(1) field lookup.

Parameters:

  • phase_id (str) – str = Phase identifier

  • query_body (str | None) – str | None = Optional GraphQL fields

Returns:

Phase = Parsed Phase model

Raises:
Example:
>>> callable(PhaseService.getPhaseModel)
True
Return type:

Phase

getPhaseField(phase_id, field_id)[source]

Retrieve phase schema metadata for a specific field.

This helper is phase-schema oriented. It does not inspect card payloads and does not depend on whether a field currently has a value in a card.

Parameters:

  • phase_id (str) – str = Phase identifier

  • field_id (str) – str = Field identifier

Returns:

PhaseField | None = Matching phase field when found

Return type:

PhaseField | None

hasPhaseField(phase_id, field_id)[source]

Check whether a field exists in a specific phase schema.

Parameters:

  • phase_id (str) – str = Phase identifier

  • field_id (str) – str = Field identifier

Returns:

bool = Whether the field exists in the phase schema

Return type:

bool

requirePhaseField(phase_id, field_id)[source]

Retrieve a phase field and fail semantically when it is missing.

Parameters:

  • phase_id (str) – str = Phase identifier

  • field_id (str) – str = Field identifier

Returns:

PhaseField = Requested phase field

Raises:

RequestError – When the field does not exist in the phase schema

Return type:

PhaseField

Pipe Service

class pipebridge.service.pipe.pipeService.PipeService(client)[source]

Bases: object

Service responsible for Pipe operations.

This service follows a layered architecture:

RAW → STRUCTURED → MODEL

It provides a unified and flexible interface to retrieve pipe data, including phases, labels, and users.

Parameters:

client (PipefyHttpClient) – PipefyHttpClient = HTTP client instance

getPipeRaw(pipe_id, query_body)[source]

Retrieve raw pipe data using a custom GraphQL query.

Parameters:

  • pipe_id (str) – str = Pipe identifier

  • query_body (str) – str = GraphQL fields to retrieve

Returns:

dict = Raw API response

Raises:
Example:
>>> callable(PipeService.getPipeRaw)
True
Return type:

Dict[str, Any]

getPipeFieldCatalog(pipe_id)[source]

Retrieve all configured phases and fields for a pipe as a model.

This method is intended for field discovery and schema inspection. It returns the canonical Pipe model, with phase fields and start form fields populated from a single pipe schema catalog query.

Parameters:

pipe_id (str) – str = Pipe identifier

Returns:

Pipe = Pipe model with phase fields populated

Raises:
Example:
>>> callable(PipeService.getPipeFieldCatalog)
True
Return type:

Pipe

getPipeStructured(pipe_id, query_body=None)[source]

Retrieve structured pipe data.

This method abstracts GraphQL complexity and returns a cleaned, consistent dictionary representing a pipe.

If no query_body is provided, a default query including phases, labels, users, and card count is used.

Parameters:

  • pipe_id (str) – str = Pipe identifier

  • query_body (str | None) – str | None = Optional GraphQL fields

Returns:

dict = Structured pipe data

Raises:
Example:
>>> callable(PipeService.getPipeStructured)
True
Return type:

Dict[str, Any]

getPipeModel(pipe_id, query_body=None)[source]

Retrieve a Pipe as a strongly-typed model.

This method converts structured pipe data into a Pipe object, enabling direct access to phases, labels, and users without loops.

Parameters:

  • pipe_id (str) – str = Pipe identifier

  • query_body (str | None) – str | None = Optional GraphQL fields

Returns:

Pipe = Parsed Pipe model

Raises:
Example:
>>> callable(PipeService.getPipeModel)
True
Return type:

Pipe

File Service

class pipebridge.service.file.fileService.FileService(client, card_service)[source]

Bases: object

Application service responsible for file operations in Pipefy.

This class acts as the orchestration layer between external interfaces (e.g., Facades) and internal domain flows.

ARCHITECTURAL ROLE:

  • Receives structured request objects (DTO pattern)

  • Delegates execution to domain flows

  • Manages dependencies and execution context

DESIGN PRINCIPLES:

  • No business logic

  • No parameter explosion

  • No data transformation logic

  • Strict delegation to flows

API DESIGN:
This service enforces a request-based API:

  • FileUploadRequest

  • FileDownloadRequest

This ensures:

  • Consistency across operations

  • Extensibility without breaking signatures

  • Strong typing and validation

Parameters:

  • client (PipefyHttpClient) – PipefyHttpClient = HTTP client instance

  • card_service (CardService) – CardService = Card service instance

Raises:

ValidationError – When dependencies are invalid

Example:
>>> from pipebridge.client.httpClient import PipefyHttpClient
>>> from pipebridge.service.card.cardService import CardService
>>> client = PipefyHttpClient("token", "https://api.pipefy.com/graphql")
>>> service = FileService(client, CardService(client))
>>> isinstance(service, FileService)
True
uploadFile(request, extra_rules=None, config=None, extra_steps_before=None, extra_steps_after=None)[source]

Uploads a file and attaches it to a Pipefy card field.

This method delegates execution to FileUploadFlow, which implements a pipeline-based upload mechanism.

Parameters:

  • request (FileUploadRequest) – FileUploadRequest = Upload request object

  • extra_rules (list[BaseRule] | None) – Optional[list[BaseRule]] = Set of custom rules constructed by the user. The rules will be applied before the upload flow.

  • config (UploadConfig | None) – Optional[UploadConfig] = Configuration for upload process

  • extra_steps_before (Sequence[BaseStep] | None) – Optional[Sequence[BaseStep]] = Additional custom steps executed before the built-in upload pipeline

  • extra_steps_after (Sequence[BaseStep] | None) – Optional[Sequence[BaseStep]] = Additional custom steps executed after the built-in upload pipeline

Returns:

FileUploadResult = Upload result metadata

Raises:

  • ValidationError – When request is invalid

  • FileFlowError – When the upload flow fails in a domain-specific way

  • WorkflowError – When rule execution or step execution fails at workflow level

  • RequestError – When a lower-level transport or API failure bubbles up

Example:
>>> callable(FileService.uploadFile)
True
Return type:

FileUploadResult

downloadAllAttachments(request)[source]

Downloads all attachments from a Pipefy card field.

This method retrieves file URLs from the specified card field, downloads each file, and saves them locally.

Parameters:

request (FileDownloadRequest) – FileDownloadRequest = Download request object

Returns:

list[Path] = List of saved file paths

Raises:
Example:
>>> callable(FileService.downloadAllAttachments)
True
Return type:

List[Path]

Models

class pipebridge.models.card.Card(id, title, pipe_id, current_phase, phases_history, fields, assignees, labels)[source]

Bases: BaseModel

Represents a Pipefy card with full structured data.

This model is the canonical source of truth for current card state inside the SDK. Internal lookup maps are maintained for efficiency, but semantic helper methods should be preferred over direct map access whenever possible.

Important semantic note about fields:

  • card.fields mirrors the fields collection returned by the Pipefy card query.

  • This collection should be treated as the set of materialized field values exposed by the API for that card.

  • It is not a complete schema catalog for the current phase or the pipe.

  • A field missing from card.fields does not imply that the field does not exist in the phase or in the pipe schema.

Preferred access patterns:

  • getField(...)

  • hasField(...)

  • requireField(...)

  • getFieldValue(...)

  • getFieldType(...)

  • getConnectorIds(...)

  • getConnectedItems(...)

  • iterFields()

Parameters:

  • id (str)

  • title (str | None)

  • pipe_id (str | None)

  • current_phase (Phase | None)

  • phases_history (List[PhaseHistory])

  • fields (List[Field])

  • assignees (List[User])

  • labels (List[Label])

classmethod fromDict(data)[source]

Safely parse a Card from structured dictionary.

This method is resilient to partial failures, null values, and inconsistent API responses. Invalid items are skipped instead of breaking the entire parsing process.

Parameters:

data (Dict[str, Any]) – dict = Structured card data

Returns:

Card = Parsed Card model

Raises:

RequestError – When input data is invalid or parsing fails critically

Example:
>>> data = {"id": "1", "fields": []}
>>> card = Card.fromDict(data)
>>> isinstance(card, Card)
True
Return type:

Card

classmethod fromResponse(response)[source]

Parse API response into Card model.

Parameters:

response (Dict[str, Any]) – dict = Raw API response

Returns:

Card

Raises:
Example:
>>> response = {"data": {"card": {"id": "1"}}}
>>> card = Card.fromResponse(response)
>>> isinstance(card, Card)
True
Return type:

Card

getField(field_id)[source]

Retrieve a field by its ID.

This is the preferred non-raising accessor for card fields. It only searches inside the card.fields payload returned by Pipefy for this card query.

Parameters:

field_id (str) – str = Field identifier

Returns:

Field | None = Field if found, otherwise None

Example:
>>> card = Card.fromDict({"id": "1", "fields": []})
>>> card.getField("x") is None
True
Return type:

Field | None

hasField(field_id)[source]

Check whether a field exists in the current card payload.

This does not validate field existence in the phase schema or pipe schema. It only answers whether the field is present inside the materialized card.fields payload returned by Pipefy.

Parameters:

field_id (str) – str = Field identifier

Returns:

bool = Whether the field exists in card.fields

Return type:

bool

requireField(field_id)[source]

Retrieve a field and fail semantically when it does not exist.

The existence check is limited to the card.fields payload already materialized for this card.

Parameters:

field_id (str) – str = Field identifier

Returns:

Field = Requested field

Raises:

ValidationError – When the field does not exist in the card

Return type:

Field

requireFieldValue(field_id)[source]

Retrieve the raw value of a field and fail when the field is missing.

The lookup is limited to the card.fields payload already materialized for this card.

Parameters:

field_id (str) – str = Field identifier

Returns:

Any = Raw field value

Raises:

ValidationError – When the field does not exist in the card

Return type:

Any

iterFields()[source]

Return all materialized card fields as an ordered list.

This list reflects the card.fields payload returned by Pipefy. It should not be treated as the complete phase or pipe schema.

Returns:

list[Field] = Card fields

Return type:

List[Field]

getFieldsByType(field_type)[source]

Retrieve all card fields matching a given type.

Parameters:

field_type (str) – str = Field type

Returns:

list[Field] = Matching fields

Return type:

List[Field]

getFieldValue(field_id)[source]

Retrieve raw value of a field.

Parameters:

field_id (str) – str = Field identifier

Returns:

Any | None = Field value

Example:
>>> card = Card.fromDict({"id": "1", "fields": []})
>>> card.getFieldValue("x") is None
True
Return type:

Any | None

getConnectorIds(field_id)[source]

Retrieve connected repo item identifiers from a connector field.

This only works when the connector field is materialized inside card.fields.

Parameters:

field_id (str) – str = Connector field identifier

Returns:

list[str] = Connected repo item identifiers

Return type:

List[str]

getConnectedItems(field_id)[source]

Retrieve connected repo item metadata from a connector field.

Parameters:

field_id (str) – str = Connector field identifier

Returns:

list[ConnectedRepoItem] = Connected item metadata

Return type:

List[Any]

hasConnectedItems(field_id)[source]

Check whether a connector field materialized connected item metadata.

Parameters:

field_id (str) – str = Connector field identifier

Returns:

bool = Whether the field exposes connected items

Return type:

bool

getConnectorValue(field_id)[source]

Retrieve a structured connector value from the current card payload.

Parameters:

field_id (str) – str = Connector field identifier

Returns:

ConnectorValue | None = Structured connector value when available

Return type:

ConnectorValue | None

getFieldType(field_id)[source]

Retrieve the field type from the current card payload.

Parameters:

field_id (str) – str = Field identifier

Returns:

str | None = Field type when available

Return type:

str | None

requireFieldType(field_id)[source]

Retrieve the field type and fail when it is unavailable.

Parameters:

field_id (str) – str = Field identifier

Returns:

str = Field type

Raises:

ValidationError – When the field is missing or has no type metadata

Return type:

str

getFieldLabel(field_id)[source]

Retrieve the field label from the current card payload.

Parameters:

field_id (str) – str = Field identifier

Returns:

str | None = Field label when available

Return type:

str | None

isFieldFilled(field_id)[source]

Check whether a field exists and contains a non-empty value.

Parameters:

field_id (str) – str = Field identifier

Returns:

bool = Whether the field is filled

Return type:

bool

getText(field_id)[source]

Retrieve a text field value.

Parameters:

field_id (str) – str = Field identifier

Returns:

str | None

Example:
>>> card = Card.fromDict({"id": "1", "fields": []})
>>> card.getText("x") is None
True
Return type:

str | None

getNumber(field_id)[source]

Retrieve a numeric field value.

Parameters:

field_id (str) – str = Field identifier

Returns:

float | None

Raises:

ValueError – When value cannot be converted to number

Example:
>>> card = Card.fromDict({"id": "1", "fields": []})
>>> card.getNumber("x") is None
True
Return type:

float | None

getBoolean(field_id)[source]

Retrieve a boolean field value.

Parameters:

field_id (str) – str = Field identifier

Returns:

bool | None

Example:
>>> card = Card.fromDict({"id": "1", "fields": []})
>>> card.getBoolean("x") is None
True
Return type:

bool | None

getDate(field_id)[source]

Retrieve a date field value (raw string).

Parameters:

field_id (str) – str = Field identifier

Returns:

str | None

Example:
>>> card = Card.fromDict({"id": "1", "fields": []})
>>> card.getDate("x") is None
True
Return type:

str | None

getLabel(label_id)[source]

Retrieve a label by its ID.

Parameters:

label_id (str) – str = Label identifier

Returns:

Label | None

Example:
>>> card = Card.fromDict({"id": "1", "labels": []})
>>> card.getLabel("x") is None
True
Return type:

Label | None

validateAgainstPhase(phase)[source]

Validate card fields against Phase schema, including type validation.

This method ensures: - required fields are present - values match expected field types

Parameters:

phase (Phase) – Phase = Phase schema

Returns:

None

Raises:

ValidationError – When required fields are missing or type validation fails

Example:
>>> card = Card.fromDict({"id": "1", "fields": []})
>>> phase = Phase.fromDict({"id": "1", "fields": []})
>>> card.validateAgainstPhase(phase)
Return type:

None

class pipebridge.models.phase.Phase(id, name=None, fields=None, cards_preview=None, cards_can_be_moved_to_phases=None, next_phase_ids=None)[source]

Bases: BaseModel

Represents a Pipefy Phase with extended metadata.

This model supports full parsing including: - basic info - fields - pipe - cards preview

This model is the canonical representation of phase schema metadata. Its internal maps exist for efficient lookup, but SDK code should prefer semantic helpers instead of direct map access whenever possible.

Preferred access patterns:

  • getField(...)

  • hasField(...)

  • requireField(...)

  • getFieldType(...)

  • getFieldOptions(...)

  • isFieldRequired(...)

  • getConnectorFields()

  • iterFields()

Parameters:

  • id (str) – str = Phase identifier

  • name (str | None) – str | None = Phase name

  • fields (List[PhaseField] | None) – list[dict] = Phase fields

  • pipe – dict | None = Parent pipe

  • cards_preview (List[Dict[str, Any]] | None) – list[dict] = Preview of cards

  • cards_can_be_moved_to_phases (List[Phase] | None)

  • next_phase_ids (List[str] | None)

property fields: List[PhaseField]

Retrieve all phase fields.

WARNING:

Prefer using getField() or [] access for O(1) lookup.

Returns:

list[PhaseField]

Example:
>>> phase = Phase.fromDict({"id": "1", "fields": []})
>>> isinstance(phase.fields, list)
True
classmethod fromDict(data)[source]

Safely parse Phase from structured data.

Parameters:

data (Dict[str, Any]) – dict = Raw phase data

Returns:

Phase

Raises:

RequestError – When parsing fails

Example:
>>> phase = Phase.fromDict({"id": "1"})
>>> isinstance(phase, Phase)
True
Return type:

Phase

getField(field_id)[source]

Retrieve a field metadata by its identifier.

This is the preferred non-raising accessor for phase schema fields.

Parameters:

field_id (str) – str = Field identifier

Returns:

PhaseField | None

Example:
>>> phase = Phase.fromDict({"id": "1", "fields": []})
>>> phase.getField("x") is None
True
Return type:

PhaseField | None

getFieldType(field_id)[source]

Retrieve the configured type of a phase field.

Parameters:

field_id (str) – str = Field identifier

Returns:

str | None = Field type when available

Return type:

str | None

requireFieldType(field_id)[source]

Retrieve the configured type of a phase field and fail if missing.

Parameters:

field_id (str) – str = Field identifier

Returns:

str = Field type

Raises:

RequestError – When the field does not exist or has no type

Return type:

str

getFieldLabel(field_id)[source]

Retrieve the configured label of a phase field.

Parameters:

field_id (str) – str = Field identifier

Returns:

str | None = Field label when available

Return type:

str | None

getFieldOptions(field_id)[source]

Retrieve the configured options of a phase field.

Parameters:

field_id (str) – str = Field identifier

Returns:

list[str] = Configured field options

Return type:

List[str]

requireFieldOptions(field_id)[source]

Retrieve the configured options of a phase field and fail if empty.

Parameters:

field_id (str) – str = Field identifier

Returns:

list[str] = Configured field options

Raises:

RequestError – When the field does not exist or has no configured options

Return type:

List[str]

isFieldRequired(field_id)[source]

Check whether a phase field is required.

Parameters:

field_id (str) – str = Field identifier

Returns:

bool = Whether the field is required

Return type:

bool

hasField(field_id)[source]

Check whether a field belongs to this phase.

Parameters:

field_id (str) – str = Field identifier

Returns:

bool = Whether the field exists

Return type:

bool

requireField(field_id)[source]

Retrieve a phase field and fail when it does not exist.

Parameters:

field_id (str) – str = Field identifier

Returns:

PhaseField = Requested field schema

Raises:

RequestError – When the field does not exist in the phase

Return type:

PhaseField

iterFields()[source]

Return all phase fields as an ordered list.

Returns:

list[PhaseField] = Phase fields

Return type:

List[PhaseField]

getMoveTargetPhase(phase_id)[source]

Retrieve an allowed destination phase by identifier.

Parameters:

phase_id (str) – str = Destination phase identifier

Returns:

Phase | None = Allowed destination phase when available

Return type:

Phase | None

canMoveToPhase(phase_id)[source]

Check whether cards from this phase can be moved to a destination.

Parameters:

phase_id (str) – str = Destination phase identifier

Returns:

bool = Whether the transition is allowed

Return type:

bool

iterMoveTargetPhases()[source]

Return all destination phases configured as valid transitions.

Returns:

list[Phase] = Allowed move target phases

Return type:

List[Phase]

getFieldsByType(field_type)[source]

Retrieve all phase fields matching a given type.

Parameters:

field_type (str) – str = Field type

Returns:

list[PhaseField] = Matching field schemas

Return type:

List[PhaseField]

getConnectorFields()[source]

Retrieve all connector fields configured in this phase.

Returns:

list[PhaseField] = Connector field schemas

Return type:

List[PhaseField]

class pipebridge.models.pipe.Pipe(id, name=None, cards_count=0, organization_id=None, start_form_fields=None, phases=None, labels=None, users=None)[source]

Bases: BaseModel

Represents a Pipefy Pipe with structured and efficient access to its related entities.

This model provides:

  • O(1) access to phases, labels, and users

  • Resilient parsing from API responses

  • Consistent interface aligned with Card and Phase models

The Pipe acts as an entry point to navigate its internal structure without requiring manual iteration. Internal maps are kept for efficient lookup, but semantic helper methods should be preferred over direct map access in SDK code.

Preferred access patterns:

  • getPhase(...)

  • hasPhase(...)

  • requirePhase(...)

  • iterPhases()

  • getStartFormField(...)

  • iterStartFormFields()

  • iterStartFormConnectorFields()

  • getLabel(...)

  • requireLabel(...)

  • iterLabels()

  • getUser(...)

  • requireUser(...)

  • iterUsers()

  • iterAllFields()

  • getFieldsByType(...)

  • getConnectorFields()

Parameters:

  • id (str) – str = Unique identifier of the pipe

  • name (str | None) – str | None = Human-readable name of the pipe

  • phases (List[Phase] | None) – list[Phase] | None = List of pipe phases

  • labels (List[Label] | None) – list[Label] | None = List of pipe labels

  • users (List[User] | None) – list[User] | None = List of pipe users

  • cards_count (int)

  • organization_id (str | None)

  • start_form_fields (List[PhaseField] | None)

Example:
>>> pipe = Pipe(id="1", name="Sales")
>>> pipe.id
'1'
classmethod fromDict(data)[source]

Create a Pipe instance from structured API data.

This method performs resilient parsing:

  • Ignores invalid nested items

  • Tracks parsing issues in _parse_errors

  • Prevents full failure due to partial inconsistencies

Parameters:

data (Dict[str, Any]) – dict = Structured pipe data

Returns:

Pipe = Parsed Pipe model

Raises:

RequestError – When input data is invalid or parsing fails critically

Example:
>>> pipe = Pipe.fromDict({"id": "1"})
>>> isinstance(pipe, Pipe)
True
Return type:

Pipe

property start_form_fields: List[PhaseField]

Retrieve all start form fields associated with the pipe.

Returns:

list[PhaseField] = Start form fields

property phases: List[Phase]

Retrieve all phases associated with the pipe.

WARNING:

Prefer using getPhase() or [] access for O(1) lookup.

Returns:

list[Phase] = List of phases

Example:
>>> pipe = Pipe.fromDict({"id": "1", "phases": []})
>>> isinstance(pipe.phases, list)
True
property labels: List[Label]

Retrieve all labels associated with the pipe.

WARNING:

Prefer using getLabel() for direct access.

Returns:

list[Label]

Example:
>>> pipe = Pipe.fromDict({"id": "1", "labels": []})
>>> isinstance(pipe.labels, list)
True
property users: List[User]

Retrieve all users associated with the pipe.

WARNING:

Prefer using getUser() for direct access.

Returns:

list[User]

Example:
>>> pipe = Pipe.fromDict({"id": "1", "users": []})
>>> isinstance(pipe.users, list)
True
getStartFormField(field_id)[source]

Retrieve a start form field by its identifier.

Parameters:

field_id (str) – str = Field identifier

Returns:

PhaseField | None = Start form field when found

Return type:

PhaseField | None

hasStartFormField(field_id)[source]

Check whether a field exists in the pipe start form.

Parameters:

field_id (str) – str = Field identifier

Returns:

bool = Whether the field exists

Return type:

bool

requireStartFormField(field_id)[source]

Retrieve a start form field and fail when it does not exist.

Parameters:

field_id (str) – str = Field identifier

Returns:

PhaseField = Requested field

Raises:

RequestError – When the field does not exist in the start form

Return type:

PhaseField

iterStartFormFields()[source]

Return all start form fields as an ordered list.

Returns:

list[PhaseField] = Start form fields

Return type:

List[PhaseField]

iterStartFormConnectorFields()[source]

Return all connector fields configured in the start form.

Returns:

list[PhaseField] = Start form connector fields

Return type:

List[PhaseField]

getPhase(phase_id)[source]

Retrieve a phase by its identifier.

Parameters:

phase_id (str) – str = Phase identifier

Returns:

Phase | None = Phase if found, otherwise None

Example:
>>> pipe = Pipe.fromDict({"id": "1", "phases": []})
>>> pipe.getPhase("x") is None
True
Return type:

Phase | None

hasPhase(phase_id)[source]

Check whether a phase exists in the pipe.

Parameters:

phase_id (str) – str = Phase identifier

Returns:

bool = Whether the phase exists

Return type:

bool

requirePhase(phase_id)[source]

Retrieve a phase and fail when it does not exist.

Parameters:

phase_id (str) – str = Phase identifier

Returns:

Phase = Requested phase

Raises:

RequestError – When the phase does not exist in the pipe

Return type:

Phase

iterPhases()[source]

Return all phases as an ordered list.

Returns:

list[Phase] = Pipe phases

Return type:

List[Phase]

iterAllFields()[source]

Return all start form and phase fields from the pipe in a flat list.

Returns:

list[Any] = All fields across start form and phases

Return type:

List[Any]

getField(field_id)[source]

Retrieve a field from the pipe schema regardless of origin.

Lookup order:

  1. start form fields

  2. phase fields

Parameters:

field_id (str) – str = Field identifier

Returns:

PhaseField | None = Matching field when found

Return type:

PhaseField | None

hasField(field_id)[source]

Check whether a field exists anywhere in the pipe schema.

Parameters:

field_id (str) – str = Field identifier

Returns:

bool = Whether the field exists

Return type:

bool

requireField(field_id)[source]

Retrieve a field from the pipe schema and fail when missing.

Parameters:

field_id (str) – str = Field identifier

Returns:

PhaseField = Requested field

Raises:

RequestError – When the field does not exist in the pipe schema

Return type:

PhaseField

getFieldsByType(field_type)[source]

Retrieve all start form and phase fields matching a given type.

Parameters:

field_type (str) – str = Field type

Returns:

list[Any] = Matching fields across start form and phases

Return type:

List[Any]

getConnectorFields()[source]

Retrieve all connector fields across start form and phases.

Returns:

list[PhaseField] = Connector field schemas

Return type:

List[PhaseField]

getLabel(label_id)[source]

Retrieve a label by its identifier.

Parameters:

label_id (str) – str = Label identifier

Returns:

Label | None

Example:
>>> pipe = Pipe.fromDict({"id": "1", "labels": []})
>>> pipe.getLabel("x") is None
True
Return type:

Label | None

hasLabel(label_id)[source]

Check whether a label exists in the pipe.

Parameters:

label_id (str) – str = Label identifier

Returns:

bool = Whether the label exists

Return type:

bool

requireLabel(label_id)[source]

Retrieve a label and fail when it does not exist.

Parameters:

label_id (str) – str = Label identifier

Returns:

Label = Requested label

Raises:

RequestError – When the label does not exist in the pipe

Return type:

Label

iterLabels()[source]

Return all labels as an ordered list.

Returns:

list[Label] = Pipe labels

Return type:

List[Label]

getUser(user_id)[source]

Retrieve a user by its identifier.

Parameters:

user_id (str) – str = User identifier

Returns:

User | None

Example:
>>> pipe = Pipe.fromDict({"id": "1", "users": []})
>>> pipe.getUser("x") is None
True
Return type:

User | None

getUserByName(user_name)[source]

Retrieve a user by human-readable name using case-insensitive matching.

Parameters:

user_name (str) – str = User display name

Returns:

User | None = Matching user when found

Return type:

User | None

requireUserByName(user_name)[source]

Retrieve a user by name and fail when it does not exist.

Parameters:

user_name (str) – str = User display name

Returns:

User = Requested user

Raises:

RequestError – When the user does not exist in the pipe user list

Return type:

User

hasUser(user_id)[source]

Check whether a user exists in the pipe user list.

Parameters:

user_id (str) – str = User identifier

Returns:

bool = Whether the user exists

Return type:

bool

requireUser(user_id)[source]

Retrieve a user and fail when it does not exist.

Parameters:

user_id (str) – str = User identifier

Returns:

User = Requested user

Raises:

RequestError – When the user does not exist in the pipe

Return type:

User

iterUsers()[source]

Return all users as an ordered list.

Returns:

list[User] = Pipe users

Return type:

List[User]

Exceptions

exception pipebridge.exceptions.PipefyError(message, class_name=None, method_name=None, *, error_code=None, context=None, cause=None, retryable=None)[source]

Bases: Exception

Base exception for all Pipefy SDK errors.

This base class keeps backward compatibility with the original class_name/method_name format while also enriching exceptions with metadata that can be used by retry policies, logging, debugging, and higher-level orchestration layers.

Parameters:

  • message (str) – str = Human-readable error description

  • class_name (Optional[str]) – Optional[str] = Origin class name

  • method_name (Optional[str]) – Optional[str] = Origin method name

  • error_code (Optional[str]) – Optional[str] = Stable machine-readable error code

  • context (Optional[Mapping[str, Any]]) – Optional[Mapping[str, Any]] = Structured error context

  • cause (Optional[Exception]) – Optional[Exception] = Original exception cause

  • retryable (Optional[bool]) – bool = Indicates whether the failure is transient

Return type:

None

default_error_code: str = 'pipebridge.error'
default_retryable: bool = False
toDict()[source]

Serialize the exception metadata to a dictionary.

Returns:

dict[str, Any] = Structured exception representation

Example:
>>> error = PipefyError("failure", error_code="test.error")
>>> error.toDict()["error_code"]
'test.error'
Return type:

dict[str, Any]

exception pipebridge.exceptions.ValidationError(message, class_name=None, method_name=None, *, error_code=None, context=None, cause=None, retryable=None)[source]

Bases: PipefyError

Base exception for validation failures.

This type represents user input or domain-state violations detected before or during SDK execution.

Parameters:

  • message (str)

  • class_name (Optional[str])

  • method_name (Optional[str])

  • error_code (Optional[str])

  • context (Optional[Mapping[str, Any]])

  • cause (Optional[Exception])

  • retryable (Optional[bool])

Return type:

None

default_error_code: str = 'validation.error'
exception pipebridge.exceptions.InvalidPhaseError(message, class_name=None, method_name=None, *, error_code=None, context=None, cause=None, retryable=None)[source]

Bases: ValidationError

Raised when a card is not in the expected phase.

Example:
>>> isinstance(InvalidPhaseError("error"), InvalidPhaseError)
True
Parameters:

  • message (str)

  • class_name (Optional[str])

  • method_name (Optional[str])

  • error_code (Optional[str])

  • context (Optional[Mapping[str, Any]])

  • cause (Optional[Exception])

  • retryable (Optional[bool])

Return type:

None

default_error_code: str = 'validation.phase.invalid'
exception pipebridge.exceptions.FieldNotInPhaseError(message, class_name=None, method_name=None, *, error_code=None, context=None, cause=None, retryable=None)[source]

Bases: ValidationError

Raised when a field does not belong to the current phase.

Example:
>>> isinstance(FieldNotInPhaseError("error"), FieldNotInPhaseError)
True
Parameters:

  • message (str)

  • class_name (Optional[str])

  • method_name (Optional[str])

  • error_code (Optional[str])

  • context (Optional[Mapping[str, Any]])

  • cause (Optional[Exception])

  • retryable (Optional[bool])

Return type:

None

default_error_code: str = 'validation.field.not_in_phase'
exception pipebridge.exceptions.InvalidFieldOptionError(message, class_name=None, method_name=None, *, error_code=None, context=None, cause=None, retryable=None)[source]

Bases: ValidationError

Raised when a field value is not within the allowed options.

Example:
>>> isinstance(InvalidFieldOptionError("error"), InvalidFieldOptionError)
True
Parameters:

  • message (str)

  • class_name (Optional[str])

  • method_name (Optional[str])

  • error_code (Optional[str])

  • context (Optional[Mapping[str, Any]])

  • cause (Optional[Exception])

  • retryable (Optional[bool])

Return type:

None

default_error_code: str = 'validation.field.invalid_option'
exception pipebridge.exceptions.RequiredFieldError(message, class_name=None, method_name=None, *, error_code=None, context=None, cause=None, retryable=None)[source]

Bases: ValidationError

Raised when a required field is missing or empty.

Example:
>>> isinstance(RequiredFieldError("error"), RequiredFieldError)
True
Parameters:

  • message (str)

  • class_name (Optional[str])

  • method_name (Optional[str])

  • error_code (Optional[str])

  • context (Optional[Mapping[str, Any]])

  • cause (Optional[Exception])

  • retryable (Optional[bool])

Return type:

None

default_error_code: str = 'validation.field.required'
exception pipebridge.exceptions.ParsingError(message, class_name=None, method_name=None, *, error_code=None, context=None, cause=None, retryable=None)[source]

Bases: PipefyError

Raised when parsing data fails.

Parameters:

  • message (str)

  • class_name (Optional[str])

  • method_name (Optional[str])

  • error_code (Optional[str])

  • context (Optional[Mapping[str, Any]])

  • cause (Optional[Exception])

  • retryable (Optional[bool])

Return type:

None

exception pipebridge.exceptions.UnexpectedResponseError(message, class_name=None, method_name=None, *, error_code=None, context=None, cause=None, retryable=None)[source]

Bases: PipefyError

Raised when API response structure is invalid.

Parameters:

  • message (str)

  • class_name (Optional[str])

  • method_name (Optional[str])

  • error_code (Optional[str])

  • context (Optional[Mapping[str, Any]])

  • cause (Optional[Exception])

  • retryable (Optional[bool])

Return type:

None

exception pipebridge.exceptions.RequestError(message, class_name=None, method_name=None, *, error_code=None, context=None, cause=None, retryable=None)[source]

Bases: PipefyError

Raised when an API request fails.

Parameters:

  • message (str)

  • class_name (Optional[str])

  • method_name (Optional[str])

  • error_code (Optional[str])

  • context (Optional[Mapping[str, Any]])

  • cause (Optional[Exception])

  • retryable (Optional[bool])

Return type:

None

exception pipebridge.exceptions.IntegrationError(message, class_name=None, method_name=None, *, error_code=None, context=None, cause=None, retryable=None)[source]

Bases: PipefyError

Raised when external integrations fail.

Parameters:

  • message (str)

  • class_name (Optional[str])

  • method_name (Optional[str])

  • error_code (Optional[str])

  • context (Optional[Mapping[str, Any]])

  • cause (Optional[Exception])

  • retryable (Optional[bool])

Return type:

None

exception pipebridge.exceptions.TransportError(message, class_name=None, method_name=None, *, error_code=None, context=None, cause=None, retryable=None)[source]

Bases: RequestError

Base exception for transport-level failures.

Parameters:

  • message (str)

  • class_name (Optional[str])

  • method_name (Optional[str])

  • error_code (Optional[str])

  • context (Optional[Mapping[str, Any]])

  • cause (Optional[Exception])

  • retryable (Optional[bool])

Return type:

None

default_error_code: str = 'transport.error'
exception pipebridge.exceptions.TimeoutRequestError(message, class_name=None, method_name=None, *, error_code=None, context=None, cause=None, retryable=None)[source]

Bases: TransportError

Raised when a network request exceeds the configured timeout.

Parameters:

  • message (str)

  • class_name (Optional[str])

  • method_name (Optional[str])

  • error_code (Optional[str])

  • context (Optional[Mapping[str, Any]])

  • cause (Optional[Exception])

  • retryable (Optional[bool])

Return type:

None

default_error_code: str = 'transport.timeout'
default_retryable: bool = True
exception pipebridge.exceptions.ConnectionRequestError(message, class_name=None, method_name=None, *, error_code=None, context=None, cause=None, retryable=None)[source]

Bases: TransportError

Raised when the SDK cannot establish or maintain a connection.

Parameters:

  • message (str)

  • class_name (Optional[str])

  • method_name (Optional[str])

  • error_code (Optional[str])

  • context (Optional[Mapping[str, Any]])

  • cause (Optional[Exception])

  • retryable (Optional[bool])

Return type:

None

default_error_code: str = 'transport.connection'
default_retryable: bool = True
exception pipebridge.exceptions.RateLimitRequestError(message, class_name=None, method_name=None, *, error_code=None, context=None, cause=None, retryable=None)[source]

Bases: TransportError

Raised when the API denies the request due to rate limiting.

Parameters:

  • message (str)

  • class_name (Optional[str])

  • method_name (Optional[str])

  • error_code (Optional[str])

  • context (Optional[Mapping[str, Any]])

  • cause (Optional[Exception])

  • retryable (Optional[bool])

Return type:

None

default_error_code: str = 'transport.rate_limit'
default_retryable: bool = True
exception pipebridge.exceptions.TransportAuthenticationError(message, class_name=None, method_name=None, *, error_code=None, context=None, cause=None, retryable=None)[source]

Bases: AuthenticationError

Raised when a transport request fails due to authentication.

Parameters:

  • message (str)

  • class_name (Optional[str])

  • method_name (Optional[str])

  • error_code (Optional[str])

  • context (Optional[Mapping[str, Any]])

  • cause (Optional[Exception])

  • retryable (Optional[bool])

Return type:

None

default_error_code: str = 'transport.auth'
exception pipebridge.exceptions.TransportInvalidTokenError(message, class_name=None, method_name=None, *, error_code=None, context=None, cause=None, retryable=None)[source]

Bases: InvalidTokenError

Raised when the provided Pipefy token is invalid.

Parameters:

  • message (str)

  • class_name (Optional[str])

  • method_name (Optional[str])

  • error_code (Optional[str])

  • context (Optional[Mapping[str, Any]])

  • cause (Optional[Exception])

  • retryable (Optional[bool])

Return type:

None

default_error_code: str = 'transport.invalid_token'
exception pipebridge.exceptions.TransportUnauthorizedError(message, class_name=None, method_name=None, *, error_code=None, context=None, cause=None, retryable=None)[source]

Bases: UnauthorizedError

Raised when the authenticated principal lacks permission.

Parameters:

  • message (str)

  • class_name (Optional[str])

  • method_name (Optional[str])

  • error_code (Optional[str])

  • context (Optional[Mapping[str, Any]])

  • cause (Optional[Exception])

  • retryable (Optional[bool])

Return type:

None

default_error_code: str = 'transport.unauthorized'
exception pipebridge.exceptions.TransportUnexpectedResponseError(message, class_name=None, method_name=None, *, error_code=None, context=None, cause=None, retryable=None)[source]

Bases: UnexpectedResponseError

Raised when a transport response cannot be interpreted safely.

Parameters:

  • message (str)

  • class_name (Optional[str])

  • method_name (Optional[str])

  • error_code (Optional[str])

  • context (Optional[Mapping[str, Any]])

  • cause (Optional[Exception])

  • retryable (Optional[bool])

Return type:

None

default_error_code: str = 'transport.unexpected_response'
exception pipebridge.exceptions.AuthenticationError(message, class_name=None, method_name=None, *, error_code=None, context=None, cause=None, retryable=None)[source]

Bases: PipefyError

Base exception for authentication failures.

This category groups all token and authorization problems raised by the SDK when communicating with Pipefy.

Parameters:

  • message (str)

  • class_name (Optional[str])

  • method_name (Optional[str])

  • error_code (Optional[str])

  • context (Optional[Mapping[str, Any]])

  • cause (Optional[Exception])

  • retryable (Optional[bool])

Return type:

None

default_error_code: str = 'auth.error'
exception pipebridge.exceptions.MissingTokenError(message, class_name=None, method_name=None, *, error_code=None, context=None, cause=None, retryable=None)[source]

Bases: AuthenticationError

Raised when an authentication token is missing.

Parameters:

  • message (str)

  • class_name (Optional[str])

  • method_name (Optional[str])

  • error_code (Optional[str])

  • context (Optional[Mapping[str, Any]])

  • cause (Optional[Exception])

  • retryable (Optional[bool])

Return type:

None

default_error_code: str = 'auth.missing_token'
exception pipebridge.exceptions.InvalidTokenError(message, class_name=None, method_name=None, *, error_code=None, context=None, cause=None, retryable=None)[source]

Bases: AuthenticationError

Raised when an authentication token is invalid.

Parameters:

  • message (str)

  • class_name (Optional[str])

  • method_name (Optional[str])

  • error_code (Optional[str])

  • context (Optional[Mapping[str, Any]])

  • cause (Optional[Exception])

  • retryable (Optional[bool])

Return type:

None

default_error_code: str = 'auth.invalid_token'
exception pipebridge.exceptions.ExpiredTokenError(message, class_name=None, method_name=None, *, error_code=None, context=None, cause=None, retryable=None)[source]

Bases: AuthenticationError

Raised when an authentication token has expired.

Parameters:

  • message (str)

  • class_name (Optional[str])

  • method_name (Optional[str])

  • error_code (Optional[str])

  • context (Optional[Mapping[str, Any]])

  • cause (Optional[Exception])

  • retryable (Optional[bool])

Return type:

None

default_error_code: str = 'auth.expired_token'
exception pipebridge.exceptions.UnauthorizedError(message, class_name=None, method_name=None, *, error_code=None, context=None, cause=None, retryable=None)[source]

Bases: AuthenticationError

Raised when the caller is authenticated but not authorized.

Parameters:

  • message (str)

  • class_name (Optional[str])

  • method_name (Optional[str])

  • error_code (Optional[str])

  • context (Optional[Mapping[str, Any]])

  • cause (Optional[Exception])

  • retryable (Optional[bool])

Return type:

None

default_error_code: str = 'auth.unauthorized'
exception pipebridge.exceptions.ConfigurationError(message, class_name=None, method_name=None, *, error_code=None, context=None, cause=None, retryable=None)[source]

Bases: PipefyError

Base exception for invalid SDK configuration.

Parameters:

  • message (str)

  • class_name (Optional[str])

  • method_name (Optional[str])

  • error_code (Optional[str])

  • context (Optional[Mapping[str, Any]])

  • cause (Optional[Exception])

  • retryable (Optional[bool])

Return type:

None

default_error_code: str = 'config.error'
exception pipebridge.exceptions.PipefyInitializationError(message, class_name=None, method_name=None, *, error_code=None, context=None, cause=None, retryable=None)[source]

Bases: ConfigurationError

Raised when SDK initialization fails.

Parameters:

  • message (str)

  • class_name (Optional[str])

  • method_name (Optional[str])

  • error_code (Optional[str])

  • context (Optional[Mapping[str, Any]])

  • cause (Optional[Exception])

  • retryable (Optional[bool])

Return type:

None

default_error_code: str = 'config.initialization'
exception pipebridge.exceptions.MissingBaseUrlError(message, class_name=None, method_name=None, *, error_code=None, context=None, cause=None, retryable=None)[source]

Bases: ConfigurationError

Raised when base_url is not provided.

Parameters:

  • message (str)

  • class_name (Optional[str])

  • method_name (Optional[str])

  • error_code (Optional[str])

  • context (Optional[Mapping[str, Any]])

  • cause (Optional[Exception])

  • retryable (Optional[bool])

Return type:

None

default_error_code: str = 'config.missing_base_url'
exception pipebridge.exceptions.WorkflowError(message, class_name=None, method_name=None, *, error_code=None, context=None, cause=None, retryable=None)[source]

Bases: PipefyError

Base exception for workflow execution failures.

Parameters:

  • message (str)

  • class_name (Optional[str])

  • method_name (Optional[str])

  • error_code (Optional[str])

  • context (Optional[Mapping[str, Any]])

  • cause (Optional[Exception])

  • retryable (Optional[bool])

Return type:

None

default_error_code: str = 'workflow.error'
exception pipebridge.exceptions.RuleExecutionError(message, class_name=None, method_name=None, *, error_code=None, context=None, cause=None, retryable=None)[source]

Bases: WorkflowError

Raised when a workflow rule fails technically during execution.

Parameters:

  • message (str)

  • class_name (Optional[str])

  • method_name (Optional[str])

  • error_code (Optional[str])

  • context (Optional[Mapping[str, Any]])

  • cause (Optional[Exception])

  • retryable (Optional[bool])

Return type:

None

default_error_code: str = 'workflow.rule_execution'
exception pipebridge.exceptions.StepExecutionError(message, class_name=None, method_name=None, *, error_code=None, context=None, cause=None, retryable=None)[source]

Bases: WorkflowError

Raised when a workflow step fails technically during execution.

Parameters:

  • message (str)

  • class_name (Optional[str])

  • method_name (Optional[str])

  • error_code (Optional[str])

  • context (Optional[Mapping[str, Any]])

  • cause (Optional[Exception])

  • retryable (Optional[bool])

Return type:

None

default_error_code: str = 'workflow.step_execution'
exception pipebridge.exceptions.RetryExhaustedError(message, class_name=None, method_name=None, *, error_code=None, context=None, cause=None, retryable=None)[source]

Bases: WorkflowError

Raised when retry attempts are exhausted.

Parameters:

  • message (str)

  • class_name (Optional[str])

  • method_name (Optional[str])

  • error_code (Optional[str])

  • context (Optional[Mapping[str, Any]])

  • cause (Optional[Exception])

  • retryable (Optional[bool])

Return type:

None

default_error_code: str = 'workflow.retry_exhausted'
default_retryable: bool = False
exception pipebridge.exceptions.CircuitBreakerOpenError(message, class_name=None, method_name=None, *, error_code=None, context=None, cause=None, retryable=None)[source]

Bases: WorkflowError

Raised when a circuit breaker blocks step execution.

Parameters:

  • message (str)

  • class_name (Optional[str])

  • method_name (Optional[str])

  • error_code (Optional[str])

  • context (Optional[Mapping[str, Any]])

  • cause (Optional[Exception])

  • retryable (Optional[bool])

Return type:

None

default_error_code: str = 'workflow.circuit_open'
default_retryable: bool = False
exception pipebridge.exceptions.PipefyAPIError(message, errors=None, class_name=None, method_name=None, *, error_code=None, context=None, cause=None, retryable=None)[source]

Bases: PipefyError

Raised when Pipefy returns an explicit API-level error payload.

Parameters:

  • message (str) – str = Error message

  • errors (Any | None) – Any | None = Raw GraphQL or API error payload

  • class_name (str | None) – str | None = Class name context

  • method_name (str | None) – str | None = Method name context

  • error_code (str | None)

  • context (dict[str, Any] | None)

  • cause (Exception | None)

  • retryable (bool | None)

Example:
>>> isinstance(PipefyAPIError("error"), PipefyAPIError)
True
Return type:

None

default_error_code: str = 'api.error'
exception pipebridge.exceptions.FileUploadError(message, class_name=None, method_name=None, *, error_code=None, context=None, cause=None, retryable=None)[source]

Bases: RequestError

Legacy-compatible base exception for upload flow failures.

Parameters:

  • message (str)

  • class_name (Optional[str])

  • method_name (Optional[str])

  • error_code (Optional[str])

  • context (Optional[Mapping[str, Any]])

  • cause (Optional[Exception])

  • retryable (Optional[bool])

Return type:

None

default_error_code: str = 'file.upload'
exception pipebridge.exceptions.FileAttachError(message, class_name=None, method_name=None, *, error_code=None, context=None, cause=None, retryable=None)[source]

Bases: FileUploadError

Raised when attaching uploaded files to a card fails.

Parameters:

  • message (str)

  • class_name (Optional[str])

  • method_name (Optional[str])

  • error_code (Optional[str])

  • context (Optional[Mapping[str, Any]])

  • cause (Optional[Exception])

  • retryable (Optional[bool])

Return type:

None

default_error_code: str = 'file.attach'
exception pipebridge.exceptions.FileValidationError(message, class_name=None, method_name=None, *, error_code=None, context=None, cause=None, retryable=None)[source]

Bases: FileUploadError

Raised when file input validation fails.

Parameters:

  • message (str)

  • class_name (Optional[str])

  • method_name (Optional[str])

  • error_code (Optional[str])

  • context (Optional[Mapping[str, Any]])

  • cause (Optional[Exception])

  • retryable (Optional[bool])

Return type:

None

default_error_code: str = 'file.validation'
exception pipebridge.exceptions.FileFlowError(message, class_name=None, method_name=None, *, error_code=None, context=None, cause=None, retryable=None)[source]

Bases: FileUploadError

Base exception for file flow failures.

Parameters:

  • message (str)

  • class_name (Optional[str])

  • method_name (Optional[str])

  • error_code (Optional[str])

  • context (Optional[Mapping[str, Any]])

  • cause (Optional[Exception])

  • retryable (Optional[bool])

Return type:

None

default_error_code: str = 'file.error'
exception pipebridge.exceptions.PresignedUrlCreationError(message, class_name=None, method_name=None, *, error_code=None, context=None, cause=None, retryable=None)[source]

Bases: FileFlowError

Raised when the SDK cannot obtain a presigned upload URL.

Parameters:

  • message (str)

  • class_name (Optional[str])

  • method_name (Optional[str])

  • error_code (Optional[str])

  • context (Optional[Mapping[str, Any]])

  • cause (Optional[Exception])

  • retryable (Optional[bool])

Return type:

None

default_error_code: str = 'file.presigned_url'
exception pipebridge.exceptions.FileTransferError(message, class_name=None, method_name=None, *, error_code=None, context=None, cause=None, retryable=None)[source]

Bases: FileFlowError

Raised when binary upload/download transfer fails.

Parameters:

  • message (str)

  • class_name (Optional[str])

  • method_name (Optional[str])

  • error_code (Optional[str])

  • context (Optional[Mapping[str, Any]])

  • cause (Optional[Exception])

  • retryable (Optional[bool])

Return type:

None

default_error_code: str = 'file.transfer'
default_retryable: bool = True
exception pipebridge.exceptions.AttachmentMergeError(message, class_name=None, method_name=None, *, error_code=None, context=None, cause=None, retryable=None)[source]

Bases: FileFlowError

Raised when attachment state cannot be merged safely.

Parameters:

  • message (str)

  • class_name (Optional[str])

  • method_name (Optional[str])

  • error_code (Optional[str])

  • context (Optional[Mapping[str, Any]])

  • cause (Optional[Exception])

  • retryable (Optional[bool])

Return type:

None

default_error_code: str = 'file.attachment_merge'
exception pipebridge.exceptions.AttachmentUpdateError(message, class_name=None, method_name=None, *, error_code=None, context=None, cause=None, retryable=None)[source]

Bases: FileAttachError

Raised when the attachment field cannot be updated in Pipefy.

Parameters:

  • message (str)

  • class_name (Optional[str])

  • method_name (Optional[str])

  • error_code (Optional[str])

  • context (Optional[Mapping[str, Any]])

  • cause (Optional[Exception])

  • retryable (Optional[bool])

Return type:

None

default_error_code: str = 'file.attachment_update'
exception pipebridge.exceptions.FileDownloadError(message, class_name=None, method_name=None, *, error_code=None, context=None, cause=None, retryable=None)[source]

Bases: RequestError

Raised when attachment download flow fails.

Parameters:

  • message (str)

  • class_name (Optional[str])

  • method_name (Optional[str])

  • error_code (Optional[str])

  • context (Optional[Mapping[str, Any]])

  • cause (Optional[Exception])

  • retryable (Optional[bool])

Return type:

None

default_error_code: str = 'file.download'
default_retryable: bool = True
pipebridge.exceptions.getExceptionContext(obj)[source]

Extract class and method name for exception context.

Parameters:

obj (object) – object = Instance

Returns:

tuple[str, str]

Return type:

Tuple[str, str]