| Name | bisslog JSON |
| Version |
0.0.9
JSON |
| download |
| home_page | None |
| Summary | Business logic core (bisslog-core) - A lightweight and dependency-free implementation of Hexagonal Architecture (Ports and Adapters) in Python. |
| upload_time | 2025-10-13 23:24:30 |
| maintainer | None |
| docs_url | None |
| author | None |
| requires_python | >=3.7 |
| license | None |
| keywords |
hexagonal
ports
adapters
business
logic
usecase
domain
ddd
|
| VCS |
 |
| bugtrack_url |
|
| requirements |
No requirements were recorded.
|
| Travis-CI |
No Travis.
|
| coveralls test coverage |
No coveralls.
|
# Business logic core for python (bisslog-core-py)
[](https://pypi.org/project/bisslog/)
[](LICENSE)
**bisslog-core** is a lightweight and dependency-free Python library that implements the **Hexagonal Architecture (Ports and Adapters)**.
It enforces a strict separation between **domain logic, application, and infrastructure**, allowing easy integration with different frameworks and external services without modifying core business logic.
This library serves as an **auxiliary layer for business logic or service domain**, providing a **common language** for operations when interacting with external components.
In other words, the **business rules remain unchanged** even if the infrastructure changes (e.g., switching the messaging system).
The key principle is:
> **"The domain should not change because some adapter changed."**
It is designed for **backend functionality**, **framework-agnostic development**, and to **minimize migration costs**.
---
## 🚀 Installation
You can install `bisslog-core` using **pip**:
```bash
pip install bisslog
```
## Usage Example
Here is an example of what the implementation of a use case looks like without importing any additional dependencies.
Note: if the adapter is not implemented it will give execution attempt messages.
~~~python
from random import random
from bisslog.database.bisslog_db import bisslog_db as db
from bisslog.use_cases.use_case_full import FullUseCase
from bisslog import use_case
from scripts.project_example_1.usecases.my_second_use_case import my_second_use_case
class SumarUseCase(FullUseCase):
@use_case # or simply def use()
def something(self, a: int, b: int, user_id: int, transaction_id: str, *args, **kwargs) -> dict:
component = self._transaction_manager.get_component()
self.log.info("Receive a:%d b:%d %s", a, b, component, checkpoint_id="reception",
transaction_id=transaction_id)
# Retrieve last session
last_session = db.session.get_last_session_user(user_id)
if last_session is not None:
self.log.info(f"Last session of user {user_id} fue {last_session}",
checkpoint_id="last_session")
db.session.save_new_session_user(user_id)
db.event_type.loadWebhookEventType(5)
rand = random()
new_value = my_second_use_case(value=rand*10, product_type="string2", transaction_id=transaction_id)
# Calculate result
res = a + b
if res > 10:
self.log.warning("Es mayor que 10", checkpoint_id="check-response")
# Publish event
self.publish("queue_suma", {"suma": res + new_value, "operation": "a + b"})
return {"suma": res}
sumar_use_case = SumarUseCase("sumar")
~~~
~~~python
from bisslog import use_case, domain_context, transaction_manager, bisslog_upload_file
log = domain_context.tracer
@use_case
def my_second_use_case(value: float, product_type: str, *args, **kwargs) -> float:
log.info(
"Received %d %s", value, transaction_manager.get_component(),
checkpoint_id="second-reception")
if product_type == "string1":
new_value = value * .2
elif product_type == "string2":
new_value = value * .3
elif product_type == "string3":
new_value = value * .5
else:
new_value = value * .05
uploaded = bisslog_upload_file.main.upload_file_from_local("./test.txt", "/app/casa/20")
if uploaded:
log.info("Uploaded file component: %s", transaction_manager.get_component(),
checkpoint_id="uploaded-file")
return new_value
~~~
For the configuration of the entry-points or primary libraries, they will only have to call the corresponding use case and map the fields. Here is an example with FastAPI. [More examples](scripts/project_example_1/)
~~~python
from typing import Annotated
from fastapi import FastAPI, Header, HTTPException
from scripts.project_example_1.usecases.my_first_use_case import sumar_use_case
app = FastAPI()
@app.get("/fast-api-example/first-use-case/{a}/{b}")
async def get_user(a: str, b: str,
user_id: Annotated[str | None, Header()]):
print(a, b, user_id)
if user_id is None:
raise HTTPException(status_code=400, detail="user_id is required")
res = sumar_use_case(int(a), int(b), int(user_id))
res["identifier"] = "fast-api"
return res
~~~
Definition of possible interfaces of divisions needed by the database, if not, it does not matter.
~~~python
from abc import ABC, abstractmethod
from typing import Optional
from bisslog.adapters.division import Division
class SessionDivision(Division, ABC):
@abstractmethod
def save_new_session_user(self, user_identifier: int) -> None:
raise NotImplementedError("save_new_session_user not implemented")
@abstractmethod
def get_last_session_user(self, user_identifier: int) -> Optional[dict]:
raise NotImplementedError("get_last_session_user not implemented")
~~~
The following is a dummie implementation of the above division to give an example
~~~python
import uuid
from datetime import datetime, timezone
from typing import Optional
from scripts.project_example_1.database.session_division import SessionDivision
cache_db = {"session": []}
class SessionDivisionCache(SessionDivision):
def get_last_session_user(self, user_identifier: int) -> Optional[dict]:
user_sessions = [session for session in cache_db["session"]
if session["user"] == user_identifier]
if not user_sessions:
return None
return user_sessions[-1]
def save_new_session_user(self, user_identifier: int) -> None:
cache_db["session"].append({"user": user_identifier, "id": uuid.uuid4().hex,
"created_at": datetime.now(timezone.utc)})
session_division_cache = SessionDivisionCache()
~~~
## 🧪 Running library tests
To Run test with coverage
~~~cmd
coverage run --source=bisslog -m pytest tests/
~~~
To generate report
~~~cmd
coverage html && open htmlcov/index.html
~~~
## 📜 License
This project is licensed under the MIT License. See the [LICENSE](LICENSE) file for details.
Raw data
{
"_id": null,
"home_page": null,
"name": "bisslog",
"maintainer": null,
"docs_url": null,
"requires_python": ">=3.7",
"maintainer_email": null,
"keywords": "hexagonal, ports, adapters, business, logic, usecase, domain, DDD",
"author": null,
"author_email": "Darwin Stiven Herrera Cartagena <darwinsherrerac@gmail.com>",
"download_url": "https://files.pythonhosted.org/packages/89/46/9843d722a20cfacdc2d5f58d01d11c20fd52f97109976ab28fbdf9bafa22/bisslog-0.0.9.tar.gz",
"platform": null,
"description": "# Business logic core for python (bisslog-core-py)\n\n[](https://pypi.org/project/bisslog/)\n[](LICENSE)\n\n**bisslog-core** is a lightweight and dependency-free Python library that implements the **Hexagonal Architecture (Ports and Adapters)**. \nIt enforces a strict separation between **domain logic, application, and infrastructure**, allowing easy integration with different frameworks and external services without modifying core business logic.\n\nThis library serves as an **auxiliary layer for business logic or service domain**, providing a **common language** for operations when interacting with external components. \nIn other words, the **business rules remain unchanged** even if the infrastructure changes (e.g., switching the messaging system). \nThe key principle is: \n> **\"The domain should not change because some adapter changed.\"**\n\nIt is designed for **backend functionality**, **framework-agnostic development**, and to **minimize migration costs**.\n\n\n\n\n\n---\n\n## \ud83d\ude80 Installation\nYou can install `bisslog-core` using **pip**:\n\n```bash\npip install bisslog\n```\n\n## Usage Example\n\n\nHere is an example of what the implementation of a use case looks like without importing any additional dependencies.\n\n\nNote: if the adapter is not implemented it will give execution attempt messages.\n\n~~~python\nfrom random import random\n\nfrom bisslog.database.bisslog_db import bisslog_db as db\nfrom bisslog.use_cases.use_case_full import FullUseCase\nfrom bisslog import use_case\nfrom scripts.project_example_1.usecases.my_second_use_case import my_second_use_case\n\n\nclass SumarUseCase(FullUseCase):\n\n @use_case # or simply def use()\n def something(self, a: int, b: int, user_id: int, transaction_id: str, *args, **kwargs) -> dict:\n component = self._transaction_manager.get_component()\n self.log.info(\"Receive a:%d b:%d %s\", a, b, component, checkpoint_id=\"reception\",\n transaction_id=transaction_id)\n\n # Retrieve last session\n last_session = db.session.get_last_session_user(user_id)\n if last_session is not None:\n self.log.info(f\"Last session of user {user_id} fue {last_session}\", \n checkpoint_id=\"last_session\")\n\n db.session.save_new_session_user(user_id)\n\n db.event_type.loadWebhookEventType(5)\n rand = random()\n new_value = my_second_use_case(value=rand*10, product_type=\"string2\", transaction_id=transaction_id)\n\n # Calculate result\n res = a + b\n if res > 10:\n self.log.warning(\"Es mayor que 10\", checkpoint_id=\"check-response\")\n\n # Publish event\n self.publish(\"queue_suma\", {\"suma\": res + new_value, \"operation\": \"a + b\"})\n return {\"suma\": res}\n\n\nsumar_use_case = SumarUseCase(\"sumar\")\n~~~\n\n~~~python\nfrom bisslog import use_case, domain_context, transaction_manager, bisslog_upload_file\n\n\nlog = domain_context.tracer\n\n@use_case\ndef my_second_use_case(value: float, product_type: str, *args, **kwargs) -> float:\n\n log.info(\n \"Received %d %s\", value, transaction_manager.get_component(),\n checkpoint_id=\"second-reception\")\n\n if product_type == \"string1\":\n new_value = value * .2\n elif product_type == \"string2\":\n new_value = value * .3\n elif product_type == \"string3\":\n new_value = value * .5\n else:\n new_value = value * .05\n\n uploaded = bisslog_upload_file.main.upload_file_from_local(\"./test.txt\", \"/app/casa/20\")\n\n if uploaded:\n\n log.info(\"Uploaded file component: %s\", transaction_manager.get_component(),\n checkpoint_id=\"uploaded-file\")\n\n return new_value\n~~~\n\n\n\nFor the configuration of the entry-points or primary libraries, they will only have to call the corresponding use case and map the fields. Here is an example with FastAPI. [More examples](scripts/project_example_1/)\n\n\n~~~python\nfrom typing import Annotated\n\nfrom fastapi import FastAPI, Header, HTTPException\n\nfrom scripts.project_example_1.usecases.my_first_use_case import sumar_use_case\n\napp = FastAPI()\n\n@app.get(\"/fast-api-example/first-use-case/{a}/{b}\")\nasync def get_user(a: str, b: str,\n user_id: Annotated[str | None, Header()]):\n print(a, b, user_id)\n if user_id is None:\n raise HTTPException(status_code=400, detail=\"user_id is required\")\n res = sumar_use_case(int(a), int(b), int(user_id))\n res[\"identifier\"] = \"fast-api\"\n return res\n\n~~~\n\n\nDefinition of possible interfaces of divisions needed by the database, if not, it does not matter.\n\n~~~python\nfrom abc import ABC, abstractmethod\nfrom typing import Optional\n\nfrom bisslog.adapters.division import Division\n\n\nclass SessionDivision(Division, ABC):\n\n @abstractmethod\n def save_new_session_user(self, user_identifier: int) -> None:\n raise NotImplementedError(\"save_new_session_user not implemented\")\n\n @abstractmethod\n def get_last_session_user(self, user_identifier: int) -> Optional[dict]:\n raise NotImplementedError(\"get_last_session_user not implemented\")\n\n~~~\n\nThe following is a dummie implementation of the above division to give an example\n\n~~~python\nimport uuid\nfrom datetime import datetime, timezone\nfrom typing import Optional\n\nfrom scripts.project_example_1.database.session_division import SessionDivision\n\n\ncache_db = {\"session\": []}\n\nclass SessionDivisionCache(SessionDivision):\n\n def get_last_session_user(self, user_identifier: int) -> Optional[dict]:\n\n user_sessions = [session for session in cache_db[\"session\"]\n if session[\"user\"] == user_identifier]\n if not user_sessions:\n return None\n return user_sessions[-1]\n\n\n def save_new_session_user(self, user_identifier: int) -> None:\n cache_db[\"session\"].append({\"user\": user_identifier, \"id\": uuid.uuid4().hex,\n \"created_at\": datetime.now(timezone.utc)})\n\nsession_division_cache = SessionDivisionCache()\n\n~~~\n\n\n## \ud83e\uddea Running library tests\n\nTo Run test with coverage\n~~~cmd\ncoverage run --source=bisslog -m pytest tests/\n~~~\n\n\nTo generate report\n~~~cmd\ncoverage html && open htmlcov/index.html\n~~~\n\n\n## \ud83d\udcdc License\n\nThis project is licensed under the MIT License. See the [LICENSE](LICENSE) file for details.\n\n",
"bugtrack_url": null,
"license": null,
"summary": "Business logic core (bisslog-core) - A lightweight and dependency-free implementation of Hexagonal Architecture (Ports and Adapters) in Python.",
"version": "0.0.9",
"project_urls": {
"Homepage": "https://github.com/darwinhc/bisslog-core-py"
},
"split_keywords": [
"hexagonal",
" ports",
" adapters",
" business",
" logic",
" usecase",
" domain",
" ddd"
],
"urls": [
{
"comment_text": null,
"digests": {
"blake2b_256": "5f9a852eb3561b1f2827ce9d0bbb0d6fcf5fc80c43c735e65d864ae21f0f5d2b",
"md5": "5360546e80b43242377737659a62c3df",
"sha256": "9f54cdb837d1de57b484724e9085eaa2a02fba2719c9eab0def91ed0dd2e0b63"
},
"downloads": -1,
"filename": "bisslog-0.0.9-py3-none-any.whl",
"has_sig": false,
"md5_digest": "5360546e80b43242377737659a62c3df",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": ">=3.7",
"size": 47599,
"upload_time": "2025-10-13T23:24:25",
"upload_time_iso_8601": "2025-10-13T23:24:25.629527Z",
"url": "https://files.pythonhosted.org/packages/5f/9a/852eb3561b1f2827ce9d0bbb0d6fcf5fc80c43c735e65d864ae21f0f5d2b/bisslog-0.0.9-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": null,
"digests": {
"blake2b_256": "89469843d722a20cfacdc2d5f58d01d11c20fd52f97109976ab28fbdf9bafa22",
"md5": "7d72638c1e147876674c9e46f196771f",
"sha256": "59d39502d2b820af6de093a20f3ef2ac96d73396bd86759e93bd6f7d0e001c72"
},
"downloads": -1,
"filename": "bisslog-0.0.9.tar.gz",
"has_sig": false,
"md5_digest": "7d72638c1e147876674c9e46f196771f",
"packagetype": "sdist",
"python_version": "source",
"requires_python": ">=3.7",
"size": 159881,
"upload_time": "2025-10-13T23:24:30",
"upload_time_iso_8601": "2025-10-13T23:24:30.841473Z",
"url": "https://files.pythonhosted.org/packages/89/46/9843d722a20cfacdc2d5f58d01d11c20fd52f97109976ab28fbdf9bafa22/bisslog-0.0.9.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2025-10-13 23:24:30",
"github": true,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"github_user": "darwinhc",
"github_project": "bisslog-core-py",
"travis_ci": false,
"coveralls": false,
"github_actions": true,
"lcname": "bisslog"
}
