BasicConnectorTutorial

Tutorial: implementing a full featured YAML database connection

The following tutorial shows how to implement a basic DatabaseEngine based full featured YAML database backend. 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 dicts will always match the structure of the database, so the Item you want to initialize with it must also match the database structure. For example:

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.

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 ..database.connection import DatabaseConnection

class YamlConnection(DatabaseConnection):
    def __init__(self, file_path, yaml_path=[]):
        """Connection to a YAML file.

        :param file_path: Path to the YAML file
        :param yaml_path: YAML path to objects considered by the connection
        """
        self.file_path = pathlib.Path(file_path)
        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 dataclasses import dataclass
from ..database.request import DatabaseSearchRequest

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

@dataclass
class YamlSearchResponse:
    path: list[str|int]
    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:

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

To do so, we use the utility library from backo.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 backo.utils.nested_data_path import find
...
from .request import YamlSearchRequest, YamlSearchResponse
from ..error import NotFoundError

class YamlConnection(DatabaseConnection):
  
    ...

    def execute_search(self, yaml_search_request):
        with open(self.file_path, "r") as yaml_database:
            database = load(yaml_database.read(), Loader=Loader)
            try:
                return YamlSearchResponse(
                    yaml_search_request.path,
                    # 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 NotFoundError(
                    f"Object with id {yaml_search_request.path} not found in YAML file {self.file_path}"
                )

Item Mapper

Now we need to define 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 backo _ids.

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

# yaml/item.py

from ..database.item import ItemMapper
from .request import YamlSearchRequest


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

    def load(self, 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:

# backo/item.py

from ..database.item import ItemMapper, DatabaseItem

...

class YamlItem(DatabaseItem):
    def __init__(
        self, item_mapper=MapByKey(), model={}
    ):
        super().__init__(id_mapper, model)

# yaml/engine.py
from ..database.engine import DatabaseEngine
from .connection import YamlConnection
from .item import YamlItem


class YamlEngine(DatabaseEngine):
    def __init__(self, file_path, yaml_path=[], database_item=YamlItem()):
        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 backo.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
{'description': 'Example user', 'gid': 12, 'name': 'pipo', '_id': '1'}

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(DatabaseCreateRequest):
    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/item.py

...

from .request import YamlSearchRequest, YamlCreateRequest

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

    ...

    def create_request(self, item_value):
        _id = self.generate_id()
        return YamlCreateRequest([_id], 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.

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

# yaml/connection.py

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

...

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

...

class YamlConnection(DatabaseConnection):
    ...

    def execute_create(self, yaml_create_request):
        database = None
        with open(self.file_path, "r") 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 + yaml_create_request.path, yaml_create_request.value)

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

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

from backo.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.
{'description': 'Example user', 'gid': 12, 'name': 'pipo', '_id': 'df8a3272-c829-4a2b-b31f-a3e18d937ce4'}

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

...

from ..database.request import DatabaseSearchRequest, DatabaseCreateRequest, DatabaseDeleteRequest

...


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


@dataclass
class YamlDeleteResponse:
    pass

Here is how to process the request using the YamlConnection:

# yaml/connection.py

...

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

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

...

class YamlConnection(DatabaseConnection):
    ...

    def execute_delete(self, yaml_delete_request):
        database = None
        with open(self.file_path, "r") 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 + yaml_delete_request.path)
        except KeyError:
            # Deleting an object that do not exist is not an error
            pass

        with open(self.file_path, "w") 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 YamlKey(IdMapper):
    ...

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

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

# example/yaml/delete.py

from backo.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.
{'description': 'Example user', 'gid': 12, 'name': 'pipo', '_id': 'a463b384-a7ee-41ee-88ae-052dde322b35'}
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

...

from ..database.request import (
    DatabaseSearchRequest,
    DatabaseCreateRequest,
    DatabaseDeleteRequest,
    DatabaseUpdateRequest,
)

...


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


@dataclass
class YamlUpdateResponse:
    pass

Process the request using the YamlConnection:

# yaml/connection.py

...

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

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

...

class YamlConnection(DatabaseConnection):
    ...

    def execute_update(self, yaml_update_request):
        database = None
        with open(self.file_path, "r") 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 + yaml_update_request.path, yaml_update_request.value)

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

Finally, enable update in the MapToKey item mapper:

# yaml/item.py
...

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

class YamlKey(IdMapper):
    ...

    def update_request(self, _id, value):
        return YamlUpdateRequest([_id], value)

The YamlEngine can now perform item updates:

# examples/yaml/update.py

from backo.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 = 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.
{'description': 'Example user', 'gid': 12, 'name': 'pipo', '_id': 'f3ef7467-3ec7-4acd-ba89-42d256736fb5'}
Item f3ef7467-3ec7-4acd-ba89-42d256736fb5 updated.
{'_id': 'f3ef7467-3ec7-4acd-ba89-42d256736fb5', 'description': 'Updated user', 'gid': 12, 'name': 'molo'}

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 dict generated by the connector, as defined by the current implementation of load in the MapByKey item mapper.

This means the schema of the Item that we want to initialize must exactly correspond to the data structure of the YAML file. This is obviously not always the case, as it might be required to connect to an already existing database with a static structure that do not match the desired Item schema.

To handle such cases, the DatabaseEngine support 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.

Home
Database Engine

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Uh oh!

BasicConnectorTutorial

Tutorial: implementing a full featured YAML database connection

Connection

Item Mapper

Convenient defaults

Item creation

Item deletion

Item update

Conclusion

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Clone this wiki locally