🧩 Guide to Creating Database Adapters in bisslog
This guide shows how to implement database adapters in bisslog-based projects using the Ports and Adapters approach to build sustainable and decoupled architectures.
📁 Project Structure
1
2
3
4
5
6
7
8
9
10
11
12
13
14
src/
├── domain/
│ └── model/
│ └── schema.py
├── infra/
│ └── database/
│ ├── schema_division.py # Port
│ └── implementations/
│ ├── vanilla_cache/
│ │ ├── schema_vanilla_cache_division.py
│ │ ├── stores_vanilla_cache_division.py
│ │ └── schema_def_version_vanilla_cache_div.py
│ └── pymongo/
│ └── schema_pymongo_division.py
🔌 Defining the Port (Division)
The port acts as an interface between your business logic and the data layer. It defines what operations are required, not how they are implemented.
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
# schema_division.py
from abc import ABC, abstractmethod
from typing import Optional, List
from src.domain.model.schema import Schema
class SchemaDivision(ABC):
@abstractmethod
def get_schema(self, schema_keyname: str) -> Optional[Schema]: ...
@abstractmethod
def get_schemas(self, params: dict) -> List[Schema]: ...
@abstractmethod
def create_schema(self, schema: Schema): ...
🍃 Implementing an Adapter (Example: PyMongo)
Concrete adapters live in infra/database/implementations.
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
# schema_pymongo_division.py
from dataclasses import asdict
from typing import Hashable, Optional, List
from bisslog_pymongo import BasicPymongoHelper, bisslog_exc_mapper_pymongo
from src.domain.model.schema import Schema
from src.infra.database.schema_division import SchemaDivision
class SchemaMongoDivision(SchemaDivision, BasicPymongoHelper):
col = "schemas"
@bisslog_exc_mapper_pymongo
def get_schema(self, schema_keyname: str) -> Optional[Schema]:
res = self.get_collection(self.col).find_one({"schema_keyname": schema_keyname})
return self.stringify_identifier(res)
@bisslog_exc_mapper_pymongo
def get_schemas(self, params: dict) -> List[Schema]:
res = self.get_collection(self.col).find(params)
return self.stringify_list_identifier(res)
@bisslog_exc_mapper_pymongo
def create_schema(self, schema: Schema) -> Hashable:
res = self.get_collection(self.col).insert_one(asdict(schema))
return res.inserted_id
🔧 Registering Adapters with bisslog
Once your adapters are implemented, register them using bisslog’s adapter registry to make them available across the application:
1
2
3
4
5
6
7
8
9
10
11
12
from bisslog import bisslog_db as db
from src.infra.database.implementations.vanilla_cache import (
StoresVanillaCacheDivision,
SchemaVanillaCacheDivision,
SchemaDefVersionVanillaCacheDiv
)
db.register_adapters(
stores=StoresVanillaCacheDivision(),
schema=SchemaVanillaCacheDivision(),
schema_version=SchemaDefVersionVanillaCacheDiv()
)
This enables dynamic port resolution and decouples domain logic from infrastructure.
✅ Best Practices
- Ports define contracts in
infra/database/, and should be pure abstract classes. - Adapters live in subfolders like
vanilla_cacheorpymongo. - Use helper classes like
BasicPymongoHelperto avoid duplicating logic. - Decorators like
@bisslog_exc_mapper_pymongoprovide consistent tracing and error mapping.
🧪 Testing Strategies
Unit testing should use mock or in-memory adapters.
1
2
3
4
5
6
7
8
from src.infra.database.implementations.mock.mock_schema_division import MockSchemaDivision
from src.domain.model.schema import Schema
def test_create_and_get_schema():
repo = MockSchemaDivision()
schema = Schema(schema_keyname="x", fields={"f": "str"})
repo.create_schema(schema)
assert repo.get_schema("x") is not None
Integration testing can use the real adapter and a test database.
📚 Additional Resources
With this guide, you can structure your database adapters in a clean, scalable, and bisslog-compliant way.