Autoschematic

Each connector, like a Terraform provider, forms an adapter to some specific remote API or service. The config file autoschematic.ron is used to specify which connectors are enabled in an Autoschematic repository. A connector is always instantiated under some prefix; a prefix is just a top-level folder used to organize connectors & their resource & output files, whether by environment, team, or any other division within a repo.

To get started with building your own Autoschematic connector in Python, you can first clone autoschematic-sh/autoschematic-connector-template-python.

By starting from the template, you can implement this protocol with some straightforward boilerplate, and hopefully end up with a usable connector for your target API or service.

The reference Python connector for this guide set is the Grafana example in autoschematic-sdk-python/examples/grafana.py. In autoschematic.ron, a local Python connector is enabled with PythonLocal:

AutoschematicConfig(
    prefixes: {
        "main": Prefix(
            connectors: [
                Connector(
                    shortname: "grafana",
                    spec: PythonLocal(
                        // Absolute and relative paths are supported.
                        // path: "/home/chef/prog/ottercorp-connector-mything/connector.py",
                        path: "connectors/grafana.py",
                    ),
                ),
                Connector(
                    shortname: "aws/s3",
                    env: {
                        "KEY": "value",
                        "SOMETHING": "else",
                    },
                    [...]
                ),
                Connector(
                    shortname: "k8s",
                    [...]
                )
            ]
        ),
        "offices/singapore": Prefix(
            [...]
        ),
        "offices/taipei": Prefix(
            [...]
        ),
    }
)

A Python connector is just a Python program that subclasses autoschematic_sdk.Connector and hands control to connector_main():

class GrafanaConnector(Connector):
    def __init__(self, name: str, prefix: str) -> None:
        self.name = name
        self.prefix = prefix
        self.base_url: str = ""
        self.session: aiohttp.ClientSession | None = None

    async def init(self) -> None:
        self.base_url = os.environ.get("GRAFANA_URL", "").rstrip("/")
        api_key = os.environ.get("GRAFANA_API_KEY", "")

        if not self.base_url:
            raise RuntimeError("GRAFANA_URL environment variable is required")
        if not api_key:
            raise RuntimeError("GRAFANA_API_KEY environment variable is required")

        self.session = aiohttp.ClientSession(
            headers={
                "Authorization": f"Bearer {api_key}",
                "Content-Type": "application/json",
            },
        )


if __name__ == "__main__":
    connector_main(GrafanaConnector)

(Note: init() in Rust is called new(), and init() was chosen as the second stage initialization step, hence the confusing naming...)

• __init__() should just do basic initialization, and not do any config validation or longer setup.
• init() is where you read config, validate credentials etc, and build e.g. client or session objects for later use.
• Store prefix on the connector. All connector methods receive repo-relative addresses such as grafana/dashboards/adm7jzw.json, not prefix/grafana/....
• Keep reusable API state on self, such as an aiohttp.ClientSession, database pool, or authenticated SDK client.

Reference files:

• autoschematic-sdk-python/examples/grafana.py

• autoschematic-sdk-python/autoschematic_sdk/types.py

• autoschematic-sdk-python/autoschematic_sdk/init.py

Next: Designing the Address Space