Skip to content

BasicConnectorTutorial

Jump to bottom
Paul Breugnot edited this page Jul 9, 2026 · 4 revisions

Tutorial: implementing a full featured YAML database connector

The following tutorial shows how to implement a minimal full featured YAML database connector. This is the required starting point of any connector. It will only support direct initialization of DatabaseItems without custom attributes. This means the generated JSON-like structures will always match the structure of the database, so the item you want to initialize with it must also match the database structure.

This might enough in following scenarios:

  • a connector to a YAML file that directly store items in the YAML structure
  • a connector to an SQL database that returns the values of all columns in a table, formatted as a JSON-like dict
  • a connector to an external API that returns GET JSON responses, and expects the input JSON-like dict as POST payload to create items.
  • a connector to an LDAP database that returns the values of all attributes in an entry, formatted as a JSON-like dict.
  • ...

However, if you need any transformation of the database result (i.e. renaming a field, search for an attribute in a field at a custom path, etc), you'll need to extend the basic connector with custom attributes, as specified in the Advanced Connector Tutorial.

Database connection

The first step is to define a DatabaseConnection to execute requests on a YAML file. The connection is initilized with a path to the YAML file we want to use as a database, and a path to the YAML structure where objects will be stored. For example, if yaml_path is ["objects"], the connection will only consider YAML objects under the objects key.

# yaml/connection.py

import pathlib

from connecto.connection import DatabaseConnection

class YamlConnection(DatabaseConnection):
    """Connection to a YAML file.

    :param file_path: Path to the YAML file
    :param yaml_path: A nested data path to the root of the collection
    within the file.
    """
    def __init__(self, file_path, yaml_path=None):
        self.file_path = pathlib.Path(file_path)
        if yaml_path is None:
            self.yaml_path = []
        else:
            self.yaml_path = yaml_path

Then we need to give to the connection the ability to search items in the YAML file. To do so, we define a new YamlSearchRequest class that will contain search parameters required to execute a search. In this case, the path to look for in the YAML file is required. We also define a convenient response object that will store results of search operations.

# yaml/request.py

from typing import Any
from dataclasses import dataclass

@dataclass
class YamlSearchRequest:
    path: list[str|int]

@dataclass
class YamlSearchResponse:
    value: Any

Remember that a request can be anything. It defines what will be passed to the execute_search method. In this case, the execute_search methods needs to:

  1. Open and read the YAML file database
  2. Fetch the object associated at the path specified in the request
  3. Raise a ItemNotFound if no item associated to the key was found

To do so, we use the utility library from connecto.utils.nested_data_path that allows to conveniently query items in a nested dict/list structure.

# yaml/connection.py

from yaml import load, Loader
from connecto.utils.nested_data_path import find
...
from connecto.request import YamlSearchRequest, YamlSearchResponse
from connecto.error import ItemNotFound

class YamlConnection(DatabaseConnection):
  
    ...

    def execute_search(self, yaml_search_request):
        with open(self.file_path, "r", encoding="utf_8") as yaml_database:
            database = load(yaml_database.read(), Loader=Loader)
            try:
                return YamlSearchResponse(
                    # The request search path is relative to the base YAML path
                    find(database, self.yaml_path + yaml_search_request.path)
                )
            except KeyError as e:
                raise ItemNotFound(request.path, self.file_path) from e

Item Mapper

Now we need features to define a YAML DatabaseItem and actually query the database. The first things to provide is an ItemMapper. Here we define an item mapper that uses YAML object keys as _ids.

Considering the definition of our previous structure, the implementation of the item mapper is trivial.

# yaml/mapper.py

from connecto.item import ItemMapper
from connecto.request import YamlSearchRequest


class MapByKey(ItemMapper):
    def search_request(self, _id):
        return YamlSearchRequest([_id])

    def load(self, _init_value, base_search_response):
        return base_search_response.value

The search_request creates a query to the item at path [_id]. The connection will intepret this path relative to the optional yaml_path, but the ItemMapper does not need to know about it.

The load method is used to initialize the item from the response of the base search_request. The value of the response has been defined in the previous execute_search implementation: it is the nested object at the specified path in the YAML structure.

Convenient defaults

Finally, we define a custom YamlItem with convenient defaults to improve user experience:

# yaml/item.py

from connecto.item import ItemMapper, DatabaseItem

...

class YamlItem(DatabaseItem):
    def __init__(self, item_mapper=MapByKey(), model=None):
        if model is None:
            super().__init__(item_mapper, {})
        else:
            super().__init__(item_mapper, model)
# yaml/engine.py
from connecto.engine import DatabaseEngine
from .connection import YamlConnection
from .item import YamlItem


class YamlEngine(DatabaseEngine):
    def __init__(self, file_path, database_item=YamlItem(), yaml_path=None):
        super().__init__(YamlConnection(file_path, yaml_path), database_item)

It's now possible to search items in a YAML file:

# examples/yaml/search.py

import yaml

from connecto.yaml.engine import YamlEngine

def init_data(path):
    with open(path, "w") as yaml_file:
        yaml_file.write(
            yaml.dump({"1": {"name": "pipo", "gid": 12, "description": "Example user"}})
        )

if __name__ == "__main__":
    init_data("test.yaml")

    yaml_engine = YamlEngine("test.yaml")
    print(yaml_engine.search("1"))
$ uv run examples/yaml/search.py
("1", {'description': 'Example user', 'gid': 12, 'name': 'pipo'})

Item creation

We will now implement features required to add new objects to the YAML database.

First, a YamlCreateRequest object is required, with the associated response containing the new ID of the created item.

# yaml/request.py

...

@dataclass
class YamlCreateRequest:
    path: list[str|int]
    value: Any
    created_id: str|None = None


@dataclass
class YamlCreateResponse:
    created_id: str

The previous MapByKey item mapper can be extended to support item creation:

# yaml/mapper.py

import copy
import uuid
...

from .request import YamlSearchRequest, YamlCreateRequest

def uuid4():
    return str(uuid.uuid4())

class MapByKey(ItemMapper):
    def __init__(self, generate_id=uuid4):
        self.generate_id = generate_id

    ...

    def created_id(self, create_response: YamlCreateResponse):
        return create_response.created_id

    def create_request(self, item_value):
        _id = self.generate_id()
        # Use a deep copy of values for the base request so that attributes can
        # modify it later without side effects
        return YamlCreateRequest([_id], copy.deepcopy(item_value), _id)

In the case of YAML, we use the generated_id method to created a unique id. In some other database engine, the id might be automatically created by the database, and returned in a custom field of the custom DatabaseCreateResponse.

The created_id method allows to retrieve the ID of the new item from the database response.

Finally, allow the YamlConnection to create new entries, generating a new ID for each item:

# yaml/connection.py

...
from connecto.utils.nested_data_path import find, update

...

from .request import (
    YamlSearchRequest,
    YamlSearchResponse,
    YamlCreateRequest,
    YamlCreateResponse,
)

...

class YamlConnection(DatabaseConnection):
    ...

    def execute_create(self, request: YamlCreateRequest):
        database = None
        with open(self.file_path, "r", encoding="utf_8") as yaml_database:
            # Init database as an empty dict if the file is empty
            database = load(yaml_database.read(), Loader=Loader) or {}

        # The path of the item to update is relative to the base YAML path
        update(
            database,
            self.yaml_path + request.path,
            request.value,
        )

        with open(self.file_path, "w", encoding="utf_8") as yaml_database:
            yaml_database.write(dump(database, Dumper=Dumper))
        return YamlCreateResponse(request.created_id)

It's now possible to create items using the YamlEngine:

# examples/yaml/create.py

from connecto.yaml.engine import YamlEngine

if __name__ == "__main__":
    yaml_engine = YamlEngine("test.yaml")
    item_id = yaml_engine.create(
        {"name": "pipo", "gid": 12, "description": "Example user"}
    )
    print(f"Item {item_id} created.")
    print(yaml_engine.search(item_id))
$ uv run examples/yaml/create.py
Item df8a3272-c829-4a2b-b31f-a3e18d937ce4 created.
('df8a3272-c829-4a2b-b31f-a3e18d937ce4', {'description': 'Example user', 'gid': 12, 'name': 'pipo'})

Item deletion

Following the same scheme as before, we define a request type and the associated response for a delete operation. The response currently does not need to return anything as no DatabaseItem process is required once delete requests have been executed.

# yaml/request.py

...


@dataclass
class YamlDeleteRequest:
    path: list[str|int]


@dataclass
class YamlDeleteResponse:
    pass

Here is how to process the request using the YamlConnection:

# yaml/connection.py

...

from connecto.utils.nested_data_path import find, update, delete

from .request import (
    ...
    YamlDeleteRequest,
    YamlDeleteResponse
)

...

class YamlConnection(DatabaseConnection):
    ...

    def execute_delete(self, request: YamlDeleteRequest):
        database = None
        with open(self.file_path, "r", encoding="utf_8") as yaml_database:
            # Init database as an empty dict if the file is empty
            database = load(yaml_database.read(), Loader=Loader) or {}

        try:
            # The path of the item to delete is relative to the base YAML path
            delete(database, self.yaml_path + request.path)
        except KeyError:
            # Deleting an object that do not exist is not an error
            pass

        with open(self.file_path, "w", encoding="utf_8") as yaml_database:
            yaml_database.write(dump(database, Dumper=Dumper))
        return YamlDeleteResponse()

Finally, define how the item mapper should create delete requests:

# yaml/item.py

...

from .request import YamlSearchRequest, YamlCreateRequest, YamlDeleteRequest


class MapByKey(IdMapper):
    ...

    def delete_request(self, _id):
        return YamlDeleteRequest([_id])

It's now possible to delete items using the YamlEngine:

# example/yaml/delete.py

from connecto.yaml.engine import YamlEngine

if __name__ == "__main__":
    yaml_engine = YamlEngine("test.yaml")
    item_id = yaml_engine.create(
        {"name": "pipo", "gid": 12, "description": "Example user"}
    )
    print(f"Item {item_id} created.")
    print(yaml_engine.search(item_id))

    yaml_engine.delete(item_id)
    print(f"Item {item_id} deleted.")
$ uv run examples/yaml/delete.py
Item a463b384-a7ee-41ee-88ae-052dde322b35 created.
('a463b384-a7ee-41ee-88ae-052dde322b35', {'description': 'Example user', 'gid': 12, 'name': 'pipo'})
Item a463b384-a7ee-41ee-88ae-052dde322b35 deleted.

Item update

The database save() operation consists in updating the attributes of an already existing item.

To perform the operation on a YAML database, the request must contain the path associated to the item to update, and the up-to-date values of the item. There is currently nothing to return from the update operation in the response.

# yaml/request.py

...


@dataclass
class YamlUpdateRequest:
    path: list[str|int]
    value: Any


@dataclass
class YamlUpdateResponse:
    pass

Process the request using the YamlConnection:

# yaml/connection.py

...

from connecto.utils.nested_data_path import find, update, delete

from .request import (
    ...
    YamlUpdateRequest,
    YamlUpdateResponse
)

...

class YamlConnection(DatabaseConnection):
    ...

    def execute_update(self, request: YamlUpdateRequest):
        database = None
        with open(self.file_path, "r", encoding="utf_8") as yaml_database:
            # Init database as an empty dict if the file is empty
            database = load(yaml_database.read(), Loader=Loader) or {}

        # The path of the item to update is relative to the base YAML path
        update(
            database,
            self.yaml_path + request.path,
            request.value,
        )

        with open(self.file_path, "w", encoding="utf_8") as yaml_database:
            yaml_database.write(dump(database, Dumper=Dumper))
        return YamlDeleteResponse()

Finally, enable update in the MapByKey item mapper:

# yaml/item.py
...

from .request import (
    YamlSearchRequest,
    YamlCreateRequest,
    YamlDeleteRequest,
    YamlUpdateRequest,
)

class MapByKey(IdMapper):
    ...

    def update_request(self, _id, value):
        # Use a deep copy of values for the base request so that attributes can
        # modify it later without side effects
        return YamlUpdateRequest([_id], copy.deepcopy(value))

The YamlEngine can now perform item updates:

# examples/yaml/update.py

from connecto.yaml.engine import YamlEngine

if __name__ == "__main__":
    yaml_engine = YamlEngine("test.yaml")
    item_id = yaml_engine.create(
        {"name": "pipo", "gid": 12, "description": "Example user"}
    )
    print(f"Item {item_id} created.")
    print(yaml_engine.search(item_id))

    item_id, item = yaml_engine.search(item_id)
    item["name"] = "molo"
    item["description"] = "Updated user"

    yaml_engine.save(item_id, item)
    print(f"Item {item_id} updated.")
    print(yaml_engine.search(item_id))
$ uv run examples/yaml/update.py
Item f3ef7467-3ec7-4acd-ba89-42d256736fb5 created.
('f3ef7467-3ec7-4acd-ba89-42d256736fb5', {'description': 'Example user', 'gid': 12, 'name': 'pipo'})
Item f3ef7467-3ec7-4acd-ba89-42d256736fb5 updated.
('f3ef7467-3ec7-4acd-ba89-42d256736fb5', {'description': 'Updated user', 'gid': 12, 'name': 'molo'})

Item selection

Item selection is used to retrieve multiple items matching a filter from the database. connecto is only responsible to perform as best as possible to convert the filter into efficient database requests, if it's possible. Indeed, the specification of the select operation states that it must at least return all items matching the filter. This means always returning all items is a valid result for any filter.

Since a YAML is not a real database with selection features, select will be implemented ignoring the filter.

First, a YamlSelectRequest object is required, with the associated response containing an {id: value} dict of found items. Each value is the value at path for each item.

# yaml/request.py

...

@dataclass
class YamlSelectRequest:
    path: list[str | int]


@dataclass
class YamlSelectResponse:
    values: dict[str | int, Any]

The execution of select requests can be implemented as follows in the YamlConnection:

...

from .request import (
    ...
    YamlSelectRequest,
    YamlSelectResponse,
)

class YamlConnection(DatabaseConnection):
    ...

    def execute_select(self, request: YamlSelectRequest):
        database = None
        with open(self.file_path, "r", encoding="utf_8") as yaml_database:
            # Init database as an empty dict if the file is empty
            database = load(yaml_database.read(), Loader=Loader) or {}

            items = {}

            # Consider items at base yaml_path
            for key, value in find(database, self.yaml_path).items():
                # Gets the value at request.path within each item
                items[key] = find(value, request.path)
            return YamlSelectResponse(items)

Note

The current implementation assumes the collection of items is organized as an {id: value} dictionnary in the YAML file, as specified by the MapByKey item mapper. If other item mappers should be supported (for example, to find item within a list using an attribute as ID), the connection should be adapted to support this is select. This can be performed with a new select request type that is configured by each item mapper to give the YamlConnection an hint on how to manage items.

The item mapper then needs to implement the following methods to support selection:

# yaml/request.py

...

class MapByKey(ItemMapper):

    ...

    def select_request(self, _item_filter):
        # Use [] as path to select the complete item at each key
        return YamlSelectRequest([])

    def load_items(self, _init_factory, base_request_response: YamlSelectResponse):
        return list(base_request_response.values.items())

    def select_response(
        self,
        _id,
        base_request_response: YamlSelectResponse,
        _attribute_responses: YamlSelectResponse,
    ):
        # Base response for the item associated to _id
        base_response = YamlSearchResponse(base_request_response.values[_id])

        return base_response, None

The load_items method is assumed to return the base initialization of all selected items in a list of (_id, value). The usage of the _init_factory if explained in the AdvandedTutorial.

The purpose of the select_response method is to build a tuple of request that will be pass to the load() method of the item associated to _id. Considering our implementation of YamlItem.load(), this must be an object with a value attribute, so we reuse the YamlSearchResponse object for convenience. This is actually a good practice, as each item is loaded as if data was returned from a search request for each item. Since no custom attribute as been defined yet, we can ignore the _attribute_responses parameter and return None as response for attributes.

Finally, the select operation can be performed as follows:

# examples/yaml/select.py

import json

from connecto.yaml import YamlEngine, YamlItem, YamlAttribute

if __name__ == "__main__":
    yaml_engine = YamlEngine(
        "test.yaml", database_item=YamlItem(model={"login": YamlAttribute(["name"])})
    )
    yaml_engine.create({"login": "pipo", "gid": 12, "description": "Example user"})
    yaml_engine.create({"login": "molo", "gid": 13, "description": "Other user"})
    print("Selected items:")
    print(json.dumps(yaml_engine.select("*"), indent=2))
$ uv run examples/yaml/select.py
Selected items:
[
  [
    "8c59f779-a6cf-44bc-be0f-7e032ff0c72f",
    {
      "description": "Other user",
      "gid": 13,
      "login": "molo"
    }
  ],
  [
    "a0aff032-0dd8-481f-bdfe-4dccc3b6a66f",
    {
      "description": "Example user",
      "gid": 12,
      "login": "pipo"
    }
  ]
]

Conclusion

The current implementation provides a full featured YAML database connection that would be enough for some real use cases. Notice that the implementation of each operation is optional: it is valid to implement a connector that only support the search operation.

However, it is not possible to customize the structure of the JSON-like structure of items: it is assumed to exactly match the structure of the item in the YAML file.

This might be limiting when connecting to an existing YAML file with a scheme that does not match the expected JSON-like structure.

To handle such cases, connecto supports the implementation of custom DatabaseAttributes that allows to specify how to map each attribute to the backend database. This is detailed in the Advanced Connector Tutorial.

Clone this wiki locally