Skip to content
Bizon Bizon

Custom Sources API

Custom sources allow you to build your own source connectors using Python.

POST /api/custom-sources/validate

Section titled “POST /api/custom-sources/validate”

Validate custom source code without saving.

{
  "code": "from bizon.source.base import AbstractSource\n\nclass MySource(AbstractSource):\n    ..."
}

Validation Checks

Section titled “Validation Checks”
  • Syntax - Valid Python syntax
  • Security - No blocked imports (os, subprocess, etc.)
  • Structure - Required methods (get_streams, check_connection, etc.)
  • Extraction - Source name, streams, config schema

Error Response

Section titled “Error Response”
{
  "valid": false,
  "errors": [
    {
      "line": 5,
      "column": 0,
      "message": "Import 'os' is not allowed for security reasons",
      "severity": "error"
    }
  ],
  "warnings": [
    {
      "line": 10,
      "column": 4,
      "message": "Consider adding type hints for better documentation",
      "severity": "warning"
    }
  ]
}

POST /api/custom-sources

Section titled “POST /api/custom-sources”

Create a new custom source.

{
  "name": "Notion API",
  "source_name": "notion",
  "code": "from bizon.source.base import AbstractSource\n...",
  "description": "Custom Notion API connector"
}

Note: Code is validated before saving. Invalid code returns 400 Bad Request.

GET /api/custom-sources

Section titled “GET /api/custom-sources”

List custom sources (metadata only, no code).

[
  {
    "id": "uuid",
    "name": "Notion API",
    "source_name": "notion",
    "streams": ["pages", "databases"],
    "description": "Custom Notion API connector",
    "created_at": "2024-01-15T10:00:00Z"
  }
]

GET /api/custom-sources/{id}

Section titled “GET /api/custom-sources/{id}”

Get custom source with code.

{
  "id": "uuid",
  "name": "Notion API",
  "source_name": "notion",
  "streams": ["pages", "databases"],
  "description": "Custom Notion API connector",
  "code": "from bizon.source.base import AbstractSource\n...",
  "created_at": "2024-01-15T10:00:00Z",
  "updated_at": null
}

PUT /api/custom-sources/{id}

Section titled “PUT /api/custom-sources/{id}”

Update a custom source.

{
  "name": "Notion API v2",
  "code": "...",
  "description": "Updated connector"
}

DELETE /api/custom-sources/{id}

Section titled “DELETE /api/custom-sources/{id}”

Delete a custom source.

Returns 204 No Content.

POST /api/custom-sources/preview

Section titled “POST /api/custom-sources/preview”

Preview records from a custom source.

{
  "source_id": "uuid",     // OR provide code directly
  "code": "...",           // Optional if source_id provided
  "stream": "pages",
  "authentication": {
    "type": "api_key",
    "params": { "token": "secret-xxx" }
  },
  "max_records": 10
}

Error Response

Section titled “Error Response”
{
  "success": false,
  "records": [],
  "total_count": 0,
  "error": "Connection failed: Invalid API key",
  "execution_time_ms": 500
}

POST /api/custom-sources/{id}/check

Section titled “POST /api/custom-sources/{id}/check”

Test connection for a saved custom source.

{
  "authentication": {
    "type": "api_key",
    "params": { "token": "secret-xxx" }
  }
}

Response:

{
  "success": true,
  "message": "Connection successful"
}

Source Code Requirements

Section titled “Source Code Requirements”

Custom sources must:

  1. Extend AbstractSource from bizon.source.base
  2. Implement required methods:
    • get_streams() - Return list of available streams
    • check_connection() - Verify credentials
    • get_records(stream) - Yield records

Example Source

Section titled “Example Source”
from typing import Iterator
from bizon.source.base import AbstractSource
from bizon.source.models import SourceRecord


class MySource(AbstractSource):
    def get_streams(self) -> list[str]:
        return ["users", "orders"]


    def check_connection(self) -> bool:
        # Verify API credentials
        return True


    def get_records(self, stream: str) -> Iterator[SourceRecord]:
        if stream == "users":
            yield SourceRecord(id="1", data={"name": "Alice"})
            yield SourceRecord(id="2", data={"name": "Bob"})

Blocked Imports

Section titled “Blocked Imports”

For security, these imports are blocked:

  • os, sys, subprocess
  • socket, urllib, requests (use provided HTTP client)
  • eval, exec, compile
  • File system operations