Connectors

Import remote datasets in two steps: validate the source API key, then kick off the import. The import runs as an async job; the SDK polls until terminal status by default.

On the CLI, pass the source key through the PICTOGRAPH_SOURCE_KEY environment variable rather than --key: a key on the command line lands in shell history and is readable by any user on the machine.

The source provider’s key (V7 token or Roboflow key) is a separate value sent in the request body. It is used only for the call that needs it and is never persisted.

Supported providers

validate

Verify the source API key and list the remote datasets available to it. No quota is consumed, and the source API key is sent only on this call.

Source: Connectors.validate

result = client.connectors.validate(
    provider="v7",
    api_key="v7_api_token_example",
)
if result.valid:
    for ds in result.datasets:
        print(ds.id, ds.name, ds.image_count)
else:
    print("invalid:", result.error)

export PICTOGRAPH_SOURCE_KEY=v7_api_token_example
pictograph connectors validate v7

curl -s -X POST "https://api.pictograph.io/api/v1/developer/connectors/validate" \
  -H "X-API-Key: $PICTOGRAPH_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{"provider": "v7", "api_key": "v7_api_token_example"}'

Returns ValidationResult

class ValidationResult(BaseModel):
    """Outcome of Connectors.validate."""
    valid: bool
    workspace: str = ''
    datasets: list[RemoteDataset] = []
    error: str | None = None

Check valid first - .datasets is populated only on success.

check_limits

Preflight whether an import fits under your plan’s image-count and storage caps, before you start it.

Source: Connectors.check_limits

check = client.connectors.check_limits(
    total_images=12500,
    estimated_size_bytes=4_000_000_000,
)
if not check.allowed:
    print("blocked by:", check.exceeded)

pictograph connectors check-limits --images 12500 --bytes 4000000000

curl -s -X POST "https://api.pictograph.io/api/v1/developer/connectors/check-limits" \
  -H "X-API-Key: $PICTOGRAPH_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{"total_images": 12500, "estimated_size_bytes": 4000000000}'

Returns LimitCheckResult

class LimitCheckResult(BaseModel):
    """Outcome of Connectors.check_limits."""
    allowed: bool
    current_images: int
    image_limit: int
    images_after_import: int
    current_storage_bytes: int
    storage_limit_bytes: int
    storage_after_import_bytes: int
    exceeded: Optional[Literal['images', 'storage', 'both']] = None

import_

Kick off the import. The trailing underscore avoids shadowing the Python import keyword. The REST endpoint returns an import_id immediately; the SDK then polls until terminal status when wait=True.

Source: Connectors.import_

job = client.connectors.import_(
    provider="v7",
    api_key="v7_api_token_example",
    datasets=[
        {"id": "ds_abc", "name": "Road signs", "slug": "road-signs"},
    ],
    wait=True,
    poll_interval=3.0,
    timeout=3600.0,
)
print(job.import_id, job.status)
for ds in job.datasets:
    print(ds.name, ds.imported, "/", job.total_images)

export PICTOGRAPH_SOURCE_KEY=v7_api_token_example

# --dataset is repeatable and resolves ids against `connectors validate`.
pictograph connectors import v7 --dataset ds_abc --dataset ds_def

# Return as soon as the job is queued instead of waiting for it.
pictograph connectors import v7 --dataset ds_abc --no-wait

curl -s -X POST "https://api.pictograph.io/api/v1/developer/connectors/import/start" \
  -H "X-API-Key: $PICTOGRAPH_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "provider": "v7",
    "api_key": "v7_api_token_example",
    "datasets": [
      {"id": "ds_abc", "name": "Road signs", "slug": "road-signs"}
    ]
  }'

Returns ImportJob

class ImportJob(BaseModel):
    """Snapshot of an import operation - totals + per-dataset breakdown."""
    import_id: str
    status: Literal['processing', 'completed', 'error', 'cancelled']
    progress: float = 0.0
    total_images: int = 0
    imported_images: int = 0
    failed_images: int = 0
    current_dataset: str = ''
    datasets: list[DatasetImportProgress] = []

The raw REST response carries import_id plus the created dataset list; poll the status endpoint below for progress.

get_import

Fetch the current state of an import job.

Source: Connectors.get_import

job = client.connectors.get_import(
    import_id="imp_7f31a904",
)
print(job.status, job.progress, job.imported_images)

pictograph connectors status imp_7f31a904

curl -s "https://api.pictograph.io/api/v1/developer/connectors/import/status/imp_7f31a904" \
  -H "X-API-Key: $PICTOGRAPH_API_KEY"

Returns ImportJob

class ImportJob(BaseModel):
    """Snapshot of an import operation - totals + per-dataset breakdown."""
    import_id: str
    status: Literal['processing', 'completed', 'error', 'cancelled']
    progress: float = 0.0
    total_images: int = 0
    imported_images: int = 0
    failed_images: int = 0
    current_dataset: str = ''
    datasets: list[DatasetImportProgress] = []

Status transitions processing to completed, or to error / cancelled.

wait_for_import

Poll an import until it reaches a terminal status. This is an SDK convenience over get_import and uses the same status endpoint.

Source: Connectors.wait_for_import

job = client.connectors.wait_for_import(
    import_id="imp_7f31a904",
    timeout=600.0,
)
print(job.status)

# No wait command - `connectors import` already blocks unless you pass
# --no-wait. Otherwise poll status until it reports a terminal state.
pictograph connectors status imp_7f31a904

# No wait endpoint - poll this one until status is completed, error or cancelled.
curl -s "https://api.pictograph.io/api/v1/developer/connectors/import/status/imp_7f31a904" \
  -H "X-API-Key: $PICTOGRAPH_API_KEY"

Returns ImportJob

class ImportJob(BaseModel):
    """Snapshot of an import operation - totals + per-dataset breakdown."""
    import_id: str
    status: Literal['processing', 'completed', 'error', 'cancelled']
    progress: float = 0.0
    total_images: int = 0
    imported_images: int = 0
    failed_images: int = 0
    current_dataset: str = ''
    datasets: list[DatasetImportProgress] = []

cancel_import

Soft-cancel an in-flight import. Already-imported images are kept; the worker stops downloading new ones and the status transitions to cancelled.

Source: Connectors.cancel_import

job = client.connectors.cancel_import(
    import_id="imp_7f31a904",
)
print(job.status)

pictograph connectors cancel imp_7f31a904 --yes

curl -s -X POST "https://api.pictograph.io/api/v1/developer/connectors/import/cancel/imp_7f31a904" \
  -H "X-API-Key: $PICTOGRAPH_API_KEY"

Returns ImportJob

class ImportJob(BaseModel):
    """Snapshot of an import operation - totals + per-dataset breakdown."""
    import_id: str
    status: Literal['processing', 'completed', 'error', 'cancelled']
    progress: float = 0.0
    total_images: int = 0
    imported_images: int = 0
    failed_images: int = 0
    current_dataset: str = ''
    datasets: list[DatasetImportProgress] = []

Annotation conversion

V7 nested directories are preserved: each item’s source path becomes the image’s virtual directory. Roboflow has no directory concept, so those imports land at the dataset root. Classes whose source provides no color are assigned a distinct palette color per dataset.

Tier caps

Imports are charged against your storage and image-count plan caps. See Credits and your plan in the web app.

Common errors