pipebridge 0.1.1

Unofficial Python SDK for Pipefy with typed models, safe workflows, and extensible integrations.

PipeBridge

Python SDK for Pipefy integration focused on:

• simple public facade
• typed models
• safe field updates
• safe phase moves
• file upload and download
• schema caching
• extensibility through rules, handlers, policies, and steps

This project was not designed as a thin GraphQL wrapper. The goal is to provide a predictable, extensible integration layer suited for real-world automation scenarios.

Documentation: https://rmcavalcante7.github.io/pipebridge/

Installation

pip install pipebridge

For development:

pip install -e .[dev]

Quick Start

from pipebridge import PipeBridge

api = PipeBridge(
  token="YOUR_TOKEN",
  base_url="https://app.pipefy.com/queries",
)

card = api.cards.get("123456789")
print(card.title)
print(card.current_phase.name if card.current_phase else None)

Public Surface

The main SDK entry point is the facade:

api = PipeBridge(token="YOUR_TOKEN", base_url="https://app.pipefy.com/queries")

Public domains:

• api.cards
• api.phases
• api.pipes
• api.files

Sub-levels when applicable:

• api.cards.raw
• api.cards.structured
• api.phases.raw
• api.phases.structured
• api.pipes.raw
• api.pipes.structured

Objects also exposed at the package top level:

• PipefyHttpClient
• CardService
• FileService
• PipeService
• PhaseService
• FileUploadRequest
• FileDownloadRequest
• UploadConfig
• CardUpdateConfig
• CardMoveConfig

Core Capabilities

1. Card, pipe, and phase retrieval

card = api.cards.get("123")
phase = api.phases.get("456")
pipe = api.pipes.get("789")

2. Pipe schema catalog

pipe = api.pipes.getFieldCatalog("789")

for phase in pipe.iterPhases():
    print(phase.name)
    for field in phase.iterFields():
        print(field.id, field.type, field.options)

This catalog is important for:

• field discovery
• update validation
• type support
• schema caching

3. Card field updates

from pipebridge import CardUpdateConfig

result = api.cards.updateFields(
  card_id="123",
  fields={
    "titulo": "Novo valor",
    "prioridade": "Alta",
  },
  expected_phase_id="456",
  config=CardUpdateConfig(
    validate_field_existence=True,
    validate_field_options=True,
    validate_field_type=True,
    validate_field_format=True,
  ),
)

The current update flow supports important field families, including:

• short and long text
• number
• currency
• email
• date
• datetime
• due_date
• time
• select
• radio
• label_select
• checklist
• assignee_select
• attachment

Important note:

• connector is out of scope for V1 by architectural decision

4. Safe phase moves

from pipebridge import CardMoveConfig

result = api.cards.moveSafely(
  card_id="123",
  destination_phase_id="999",
  expected_current_phase_id="456",
  config=CardMoveConfig(validate_required_fields=True),
)

This flow validates:

• whether the current phase matches the expected phase, when provided
• whether the transition is allowed by the current phase configuration
• whether required fields in the destination phase are filled

5. File upload and download

from pipebridge import FileUploadRequest, FileDownloadRequest, UploadConfig

upload_request = FileUploadRequest(
  file_name="arquivo.txt",
  file_bytes=b"conteudo",
  card_id="123",
  field_id="anexos",
  organization_id="999",
  expected_phase_id="456",
)

upload_result = api.files.uploadFile(upload_request)

download_request = FileDownloadRequest(
  card_id="123",
  field_id="anexos",
  output_dir="./downloads",
)

files = api.files.downloadAllAttachments(download_request)

Extensibility

One of the project's core goals is to allow extension without forking the SDK.

1. Custom rules

You can inject extra rules into public flows.

Example with updates:

from pipebridge.exceptions import ValidationError
from pipebridge.workflow.rules.baseRule import BaseRule


class UppercaseOnlyRule(BaseRule):
  def __init__(self, field_id: str) -> None:
    self.field_id = field_id

  def execute(self, context) -> None:
    value = context.request.fields.get(self.field_id)
    if not isinstance(value, str) or value != value.upper():
      raise ValidationError(
        message=f"Field '{self.field_id}' must be uppercase",
        class_name=self.__class__.__name__,
        method_name="execute",
      )


api.cards.updateField(
  card_id="123",
  field_id="codigo",
  value="VALOR",
  extra_rules=[UppercaseOnlyRule("codigo")],
)

2. Ready-to-use regex for field validation

from pipebridge.service.card.flows.update.rules.regexFieldPatternRule import (
  RegexFieldPatternRule,
)

api.cards.updateField(
  card_id="123",
  field_id="codigo",
  value="ABC-123",
  extra_rules=[
    RegexFieldPatternRule({"codigo": r"^[A-Z]{3}-\d{3}$"})
  ],
)

3. Custom update handlers

You can override or add type support at runtime:

from pipebridge.service.card.flows.update.dispatcher.baseCardFieldUpdateHandler import (
  BaseCardFieldUpdateHandler,
)
from pipebridge.service.card.flows.update.dispatcher.resolvedFieldUpdate import (
  ResolvedFieldUpdate,
)


class UppercaseTextHandler(BaseCardFieldUpdateHandler):
  def resolve(self, field_id, field_type, input_value, current_field=None, phase_field=None):
    return ResolvedFieldUpdate(
      field_id=field_id,
      field_type=field_type,
      input_value=input_value,
      current_field=current_field,
      phase_field=phase_field,
      new_value=str(input_value).strip().upper(),
    )


api.cards.updateField(
  card_id="123",
  field_id="titulo",
  value="meu texto",
  extra_handlers={"short_text": UppercaseTextHandler()},
)

4. Retry and circuit breaker policies

from pipebridge import UploadConfig
from pipebridge.workflow.config.retryConfig import RetryConfig
from pipebridge.workflow.config.circuitBreakerConfig import CircuitBreakerConfig

config = UploadConfig(
  retry=RetryConfig(max_retries=5, base_delay=1.0),
  circuit=CircuitBreakerConfig(failure_threshold=5, recovery_timeout=5.0),
)

api.files.uploadFile(request=upload_request, config=config)

5. Custom upload steps

In V1, steps extensibility is publicly exposed only for uploads:

• extra_steps_before
• extra_steps_after

from pipebridge.workflow.steps.baseStep import BaseStep


class RegisterMetadataStep(BaseStep):
  def execute(self, context) -> None:
    context.metadata["source"] = "custom-step"


api.files.uploadFile(
  request=upload_request,
  extra_steps_before=[RegisterMetadataStep()],
)

Note:

• card updates and safe moves do not yet expose custom steps in the V1 public API

Models and Semantic Navigation

SDK models were designed for semantic navigation. The goal is to avoid direct structural map access whenever possible.

Examples:

card = api.cards.get("123")

if card.hasField("titulo"):
    print(card.requireFieldValue("titulo"))

phase = api.phases.get("456")
print(phase.getFieldType("prioridade"))
print(phase.getFieldOptions("prioridade"))
print(phase.isFieldRequired("prioridade"))

pipe = api.pipes.getFieldCatalog("789")
for field in pipe.getFieldsByType("select"):
    print(field.id, field.label)

Schema Cache

The SDK provides in-memory cache for pipe schema:

• keyed by pipe_id
• with TTL
• with per-key locking
• lazy refresh on demand
• no background thread in V1

On the card facade:

stats = api.cards.getSchemaCacheStats()
entry = api.cards.getSchemaCacheEntryInfo("789")
api.cards.invalidateSchemaCache("789")

Ready-to-Use Examples

The useCases folder is the recommended starting point for end users.

It contains executable examples for:

• pipe field catalog inspection
• cascading inspection across pipes, phases, and cards
• card field updates
• updates with extra rules
• custom handler
• safe moves
• upload and download
• uploads with rules and policies
• uploads with custom steps

See useCases/README.md.

HTML Documentation

The project also includes a Sphinx documentation structure in docs/.

This is the intended path for the SDK's navigable HTML documentation, including:

• overview
• quick start
• extensibility
• API reference
• development guides

To generate locally:

pip install -e .[docs]
sphinx-build -b html docs docs/_build/html

Main documentation entry point in the repository:

• docs/index.rst

Expected URL for published documentation via GitHub Pages:

• https://rmcavalcante7.github.io/pipebridge/

Tests

The project is organized as follows:

• tests/unit
• tests/functional
• tests/integration
• useCases/

Role of each:

• unit

  • isolated logic
  • no network
  • no credentials

• functional

  • public API
  • no real Pipefy
  • with fakes/doubles

• integration

  • real Pipefy operations
  • depend on:
    • PIPEFY_API_TOKEN
    • optional PIPEFY_BASE_URL

Commands:

python -m pytest tests/unit tests/functional -v
python -m pytest tests/integration -v
python -m pytest tests -v

For real integration:

$env:PIPEFY_API_TOKEN="YOUR_TOKEN"
$env:PIPEFY_BASE_URL="https://app.pipefy.com/queries"
python -m pytest tests/integration -v

Current V1 Status

V1 is complete with:

• coherent public facade
• card update flow
• safe move flow
• upload/download flow
• semantic exceptions
• schema cache
• structured pytest suite
• end-user usage examples

Out of scope for V1:

• connector as a complete relational operation
• public steps extensibility in updates and moves

Author

Rafael Mota Cavalcante

• GitHub: rmcavalcante7
• LinkedIn: rafael-cavalcante-dev-specialist
• E-mail: [e-mail]

License

MIT