Me gustaría compartir un punto de partida, una plantilla, para interactuar con la base de datos. El código esta adaptado a un SQLite por medio de Python. Aunque se puede extrapolar a cualquier base de datos o lenguaje.
La necesidad surge porque estoy trabajando en un proyecto con arquitectura limpia. El enfoque es separar la lógica de negocio de la lógica de acceso a datos para evitar dependencias.
Por ejemplo clásico, imaginemos que yo quiero obtener un cliente de la base de datos. En lugar de hacerlo directamente, lo hago a través de una interfaz.
repo = SQLiteRepo()
response = repo.fetch_one({'table': 'client', 'filters': [('name', FilterTypes.EQUALS, 'Company')]})
# {'type': ResponseTypes.SUCCESS, 'errors': [], 'data': [{'id': 1, 'name': 'Company'}]}
Ahora resulta que encontramos problemas de concurrencia, o cualquier otro tema, y necesitamos escalar a otra base de datos diferente. Será tan sencillo como cambiar la clase.
repo = MongoRepo()
response = repo.fetch_one({'table': 'client', 'filters': [('name', FilterTypes.EQUALS, 'Company')]})
# {'type': ResponseTypes.SUCCESS, 'errors': [], 'data': [{'id': 1, 'name': 'Company'}]}
No necesitamos cambiar el código de negocio. Las funciones de la clase SQLiteRepo y MongoRepo serán las mismas, al igual que sus parámetros y sus salidas. Además, de regalo, esta estructura nos permite realizar pruebas unitarias de forma sencilla.
Bajo el mismo contexto, te presento una clase que he sido diseñando por medio de la experiencia y algunos libros de referencia.
import os
from pathlib import Path
class ResponseTypes:
SUCCESS = 'Success' # The process ended correctly
PARAMETERS_ERROR = 'ParametersError' # Missing or invalid parameters
RESOURCE_ERROR = 'ResourceError' # The process ended correctly but the resource is not available (DB, file, etc)
SYSTEM_ERROR = (
'SystemError' # The process ended with an error. Python error
)
class FilterTypes:
EQUALS = '=='
NOT_EQUALS = '!='
GREATER_THAN = '>'
LESS_THAN = '<'
GREATER_THAN_OR_EQUAL = '>='
LESS_THAN_OR_EQUAL = '<='
LIKE = 'like'
IN = 'in'
NOT_IN = 'not in'
IS_NULL = 'is null'
IS_NOT_NULL = 'is not null'
class RelationTypes:
ONE_TO_ONE = 'one_to_one'
ONE_TO_MANY = 'one_to_many'
MANY_TO_MANY = 'many_to_many'
class SQLiteRepo:
PATH = os.path.join('/', 'usr', 'src', 'app', 'db')
NAME = 'database.sqlite'
def __init__(self):
if not Path(self.__absolute_path()).exists():
self.__create_db()
self.__create_tables()
self.__add_initial_data()
def __absolute_path(self):
return os.path.join(self.PATH, self.NAME)
def __create_db(self):
pass
def __create_tables(self):
pass
def __create_initial_data(self):
pass
def __fetch_data(self, params: dict[str, dict[str, str | list]], fetch_one: bool):
"""
Fetch data from the database
:param params:
- table: The table to fetch from
- columns[]: The columns to fetch. Default is all columns.
- filters[]: The filters to apply. Default is no filters. Example: [('name', FilterTypes.EQUALS, 'Company')]
- joins[]: The relations to join. Default is no joins.
:param fetch_one: Whether to fetch one or all rows
:return: dict with the result of the query
"""
return {
'type': ResponseTypes.SUCCESS,
'errors': [],
'data': [],
}
def fetch_one(self, params: dict[str, dict[str, str | list]]) -> dict:
"""
Get one row from the database
Check the fetch_data method for more information
:param params:
- table: The table to fetch from
- columns[]: The columns to fetch. Default is all columns.
- filters[]: The filters to apply. Default is no filters. Example: [('name', FilterTypes.EQUALS, 'Company')]
- joins[]: The relations to join. Default is no joins.
:return: dict with the result of the query
"""
return self.__fetch_data(params, fetch_one=True)
def fetch_all(self, params: dict[str, dict[str, str | list]]) -> dict:
"""
Get all rows from the database
Check the fetch_data method for more information
:param params:
- table: The table to fetch from
- columns[]: The columns to fetch. Default is all columns.
- filters[]: The filters to apply. Default is no filters. Example: [('name', FilterTypes.EQUALS, 'Company')]
- joins[]: The relations to join. Default is no joins.
:return: dict with the result of the query
"""
return self.__fetch_data(params, fetch_one=False)
def insert(self, params: dict[str, dict[str, str | list]]) -> dict:
"""
Insert data into the database
Example: repo.insert({'table': 'client', 'data': {'name': 'Company'}})
:param params:
- table: The table to insert into
- data {}: The data to insert
"""
return {
'type': ResponseTypes.SUCCESS,
'errors': [],
'data': [],
}
def update(self, params: dict[str, dict[str, str | list]]) -> dict:
"""
Update data in the database
:param params:
- table: The table to update
- data {}: The data to update
- filters []: The filters to apply. Example: [('name', FilterTypes.EQUALS, 'Company')]
"""
return {
'type': ResponseTypes.SUCCESS,
'errors': [],
'data': [],
}
def delete(self, params: dict[str, dict[str, str | list]]) -> dict:
"""
Delete data from the database
:param params:
- table: The table to delete from
- filters []: The filters to apply. Example: [('name', FilterTypes.EQUALS, 'Company')]
"""
return {
'type': ResponseTypes.SUCCESS,
'errors': [],
'data': [],
}
def add_relation(self, params: dict[str, dict[str, str | list]]) -> dict:
"""
Add an association between two tables.
Example 1:1
repo.add_relation({'table_parent': 'client', 'table_child': 'address', 'relation': RelationTypes.ONE_TO_ONE}, 'foreign_key': 'client_id', 'primary_key': 'id', 'filters': [('name', FilterTypes.EQUALS, 'Company')]})
Example 1:n
repo.add_relation({'table_parent': 'client', 'table_child': 'order', 'relation': RelationTypes.ONE_TO_MANY}, 'foreign_key': 'client_id', 'primary_key': 'id', 'filters': [('name', FilterTypes.EQUALS, 'Company')]})
Example n:n
repo.add_relation({'table_parent': 'client', 'table_child': 'product', 'table_relation': 'client_product', 'relation': RelationTypes.MANY_TO_MANY, 'foreign_key': 'client_id', 'primary_key': 'id', 'filters': [('name', FilterTypes.EQUALS, 'Company')], 'data': {'name': 'Product 1'}})
:param params:
- table_parent: The parent table
- table_child: The child table
- table_relation: The relation table (for many to many relations)
- relation: The type of relation. Example: RelationTypes.ONE_TO_ONE, RelationTypes.ONE_TO_MANY, RelationTypes.MANY_TO_MANY
- primary_key: The primary key in the parent table
- foreign_key: The foreign key in the child table
- filters []: The filters to apply. Example: [('name', FilterTypes.EQUALS, 'Company')]
- data {}: The data to insert (for many to many relations)
:return: dict with the result of the query
"""
return {
'type': ResponseTypes.SUCCESS,
'errors': [],
'data': [],
}
def remove_relation(self, params: dict[str, dict[str, str | list]]) -> dict:
"""
Remove an association between two tables.
Example 1:1
repo.remove_relation({'table_parent': 'client', 'table_child': 'address', 'relation': RelationTypes.ONE_TO_ONE}, 'foreign_key': 'client_id', 'primary_key': 'id', 'filters': [('name', FilterTypes.EQUALS, 'Company')]})
Example 1:n
repo.remove_relation({'table_parent': 'client', 'table_child': 'order', 'relation': RelationTypes.ONE_TO_MANY}, 'foreign_key': 'client_id', 'primary_key': 'id', 'filters': [('name', FilterTypes.EQUALS, 'Company')]})
Example n:n
repo.remove_relation({'table_parent': 'client', 'table_child': 'product', 'table_relation': 'client_product', 'relation': RelationTypes.MANY_TO_MANY, 'foreign_key': 'client_id', 'primary_key': 'id', 'filters': [('name', FilterTypes.EQUALS, 'Company')]})
:param params:
- table_parent: The parent table
- table_child: The child table
- table_relation: The relation table (for many to many relations)
- relation: The type of relation. Example: RelationTypes.ONE_TO_ONE, RelationTypes.ONE_TO_MANY, RelationTypes.MANY_TO_MANY
- primary_key: The primary key in the parent table
- foreign_key: The foreign key in the child table
- filters []: The filters to apply. Example: [('name', FilterTypes.EQUALS, 'Company')]
:return: dict with the result of the query
"""
return {
'type': ResponseTypes.SUCCESS,
'errors': [],
'data': [],
}
¿Donde esta el código de las funciones? Lo he omitido por temas laborales. Me encantaría compartirlo pero no es posible en este momento.
Sin embargo, rellenar los huecos no te llevará mucho tiempo. Y realizar los test unitarios tampoco.
También podrías añadir una base abstracta. De esta forma, podrías añadir más repositorios con otras bases de datos.
from abc import ABC, abstractmethod
class IRepository(ABC):
"""
Interface for a database repository.
"""
@abstractmethod
def fetch_one(self, params: dict[str, dict[str, str | list]]) -> dict:
pass
@abstractmethod
def fetch_all(self, params: dict[str, dict[str, str | list]]) -> dict:
pass
@abstractmethod
def insert(self, params: dict[str, dict[str, str | list]]) -> dict:
pass
@abstractmethod
def update(self, params: dict[str, dict[str, str | list]]) -> dict:
pass
@abstractmethod
def delete(self, params: dict[str, dict[str, str | list]]) -> dict:
pass
@abstractmethod
def add_relation(self, params: dict[str, dict[str, str | list]]) -> dict:
pass
@abstractmethod
def remove_relation(self, params: dict[str, dict[str, str | list]]) -> dict:
pass
A continuación deberás aplicarla a la clase SQLiteRepo.
class SQLiteRepo(IRepository):
...
Espero que te resulte útil.
Si quieres saber más sobre como adaptar tu proyecto en Python a una arquitectura limpia, te recomiendo que leas el artículo.
{{ comments.length }} comentarios