-
Notifications
You must be signed in to change notification settings - Fork 0
BasicConnectorTutorial
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.
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_pathThen 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
@dataclass
class YamlSearchRequest:
path: list[str|int]
@dataclass
class YamlSearchResponse:
path: list[str|int]
value: AnyRemember 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
NotFoundErrorif 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}"
)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.valueThe 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.
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'}
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: strThe previous MapByKey item mapper can be extended to support item creation:
# yaml/item.py
import copy
...
from .request import YamlSearchRequest, YamlCreateRequest
class MapByKey(ItemMapper):
def __init__(self, generate_id=uuid4):
self.generate_id = generate_id
...
def create_request(self, 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(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'}
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:
passHere 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.
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:
passProcess 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):
# 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 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'}
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.