| Name | django-event-observer JSON |
| Version |
2024.1.1
JSON |
| download |
| home_page | None |
| Summary | None |
| upload_time | 2024-10-21 18:56:21 |
| maintainer | None |
| docs_url | None |
| author | None |
| requires_python | ~=3.13 |
| license | MIT License Copyright (c) 2024 Sergey Romanchuk Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the "Software"), to deal in the Software without restriction, including without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to permit persons to whom the Software is furnished to do so, subject to the following conditions: The above copyright notice and this permission notice shall be included in all copies or substantial portions of the Software. THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. |
| keywords |
|
| VCS |
 |
| bugtrack_url |
|
| requirements |
No requirements were recorded.
|
| Travis-CI |
No Travis.
|
| coveralls test coverage |
No coveralls.
|
# django_event_observer
django_event_observer is a fully-fledged Django application designed for observing and handling events within your project. It provides a flexible and extensible framework for subscribing to various event types and reacting to them accordingly.
Installation
Install the package using pip:
pip install django_event_observer
Configuration
After installation, add django_event_observer to the INSTALLED_APPS in your Django settings file:
INSTALLED_APPS = [
# ...
'django_event_observer',
# ...
]
Default Settings
By default, the DJANGO_EVENT_OBSERVER settings are as follows:
from types import MappingProxyType
DJANGO_EVENT_OBSERVER = MappingProxyType(
{
"EVENT_MANAGER": "django_event_observer.event_manager.EventManager",
"AUTO_OBSERVERS": [],
}
)
You can override the DJANGO_EVENT_OBSERVER variable in your project’s settings to specify a custom import path for the EventManager or to add automatic observers.
Customizing EventManager
If you need to use a custom EventManager, specify the import path in your settings:
DJANGO_EVENT_OBSERVER = {
"EVENT_MANAGER": "my_app.event_manager.CustomEventManager",
"AUTO_OBSERVERS": [],
}
Automatic Observers
The AUTO_OBSERVERS parameter is a list of dictionaries, each describing an observer. These observers are classes inheriting from django_event_observer.observers.abstract.base.BaseAbstractObserver.
Syntax:
```python
DJANGO_EVENT_OBSERVER = {
"EVENT_MANAGER": "django_event_observer.event_manager.EventManager",
"AUTO_OBSERVERS": [
{
"CLASS": "my_module.observers.MyObserver",
"INIT_ARGS": ["arg1", "arg2"],
"INIT_KWARGS": {"key1": "value1", "key2": "value2"},
"EVENT_TYPES": ["event_type_1", "event_type_2"],
},
# Other observers...
],
}
```
• "CLASS": A string with the import path to the observer class.
• "INIT_ARGS": A list of positional arguments for class initialization.
• "INIT_KWARGS": A dictionary of keyword arguments for class initialization.
• "EVENT_TYPES": A list of strings representing event types that the observer should subscribe to.
##Usage
Creating a Custom Observer
To create a custom observer, inherit from BaseAbstractObserver and implement the update method:
# my_module/observers.py
```python
from django_event_observer.observers.abstract.base import BaseAbstractObserver
from django_event_observer.events.base import BaseEvent
class MyObserver(BaseAbstractObserver):
def __init__(self, *args, **kwargs):
super().__init__()
# Additional initialization...
def update(self, event: BaseEvent) -> None:
# Logic to handle the event
print(f"Received event: {event}")
```
Definition of BaseAbstractObserver:
# django_event_observer/observers/abstract/base.py
from abc import ABC, abstractmethod
from ...events.base import BaseEvent
class BaseAbstractObserver(ABC):
@abstractmethod
def update(self, event: BaseEvent) -> None:
...
Any custom observer must inherit from BaseAbstractObserver and implement the update method, which will be called when an event occurs.
Creating Custom Events
Events should inherit from BaseEvent and can use generics for typing event_data:
# my_module/events.py
from django_event_observer.events.base import BaseEvent
from typing import Generic, TypeVar
T = TypeVar('T')
class MyEvent(BaseEvent[T], Generic[T]):
def __init__(self, event_type: str, event_data: T):
super().__init__(event_type, event_data)
# Additional initialization...
def __str__(self):
return f"{self.event_type}: {self.event_data}"
Definition of BaseEvent:
# django_event_observer/events/base.py
import typing
from django.apps import apps
if typing.TYPE_CHECKING:
from ..event_manager import EventManager
_EventDataT = typing.TypeVar("_EventDataT")
class BaseEvent(typing.Generic[_EventDataT]):
def __init__(self, event_type: str, event_data: _EventDataT):
self._event_manager = apps.get_app_config("django_event_observer").event_manager
self.event_type = event_type
self.event_data = event_data
def __call__(self):
self.notify()
def notify(self):
self._event_manager.notify(self)
def __str__(self):
return f"{self.event_type}: {self.event_data}"
Accessing EventManager
To subscribe or unsubscribe an observer from specific event types, you need to access the initialized EventManager. It is recommended to do this using the Django applications registry:
from django.apps import apps
event_manager = apps.get_app_config("django_event_observer").event_manager
Subscribing to Events
To subscribe an observer to a specific event type:
from my_module.observers import MyObserver
from django.apps import apps
# Get the EventManager instance
event_manager = apps.get_app_config("django_event_observer").event_manager
# Create an observer instance
observer = MyObserver()
# Subscribe the observer to an event type
event_manager.subscribe("my_event_type", observer)
Unsubscribing from Events
To unsubscribe an observer from a specific event type:
# Unsubscribe the observer from the event type
event_manager.unsubscribe("my_event_type", observer)
Sending Events
To send an event and notify all subscribed observers, create an event instance and call the notify method:
from my_module.events import MyEvent
# Create an event instance
event = MyEvent(event_type="my_event_type", event_data={"key": "value"})
# Notify observers
event.notify()
Or you can call the event instance as a function thanks to the __call__ method:
# Notify observers by calling the event
event()
You can also use the EventManager directly to notify:
# Notify observers via EventManager
event_manager.notify(event)
Full Usage Example
# observers.py
from django_event_observer.observers.abstract.base import BaseAbstractObserver
from django_event_observer.events.base import BaseEvent
class MyObserver(BaseAbstractObserver):
def update(self, event: BaseEvent) -> None:
print(f"Observer received event: {event}")
# events.py
from django_event_observer.events.base import BaseEvent
class UserRegisteredEvent(BaseEvent[dict]):
def __init__(self, user_data: dict):
super().__init__(event_type="user_registered", event_data=user_data)
def __str__(self):
return f"UserRegisteredEvent: {self.event_data}"
# main.py
from django.apps import apps
from my_module.observers import MyObserver
from my_module.events import UserRegisteredEvent
# Get the EventManager instance
event_manager = apps.get_app_config("django_event_observer").event_manager
# Create an observer instance
observer = MyObserver()
# Subscribe the observer to the "user_registered" event
event_manager.subscribe("user_registered", observer)
# Create a user registration event
user_data = {"username": "john_doe", "email": "john@example.com"}
event = UserRegisteredEvent(user_data)
# Notify observers
event.notify()
# Or call the event as a function
event()
# Unsubscribe the observer when no longer needed
event_manager.unsubscribe("user_registered", observer)
Internal Structure of EventManager
The standard EventManager looks like this:
# django_event_observer/event_manager.py
from collections import defaultdict
import typing
from .observers.abstract.base import BaseAbstractObserver
from .events.base import BaseEvent
_ObserversT = dict[str, list[BaseAbstractObserver]]
class EventManager:
_subscriptions: typing.ClassVar[_ObserversT] = defaultdict(list)
def subscribe(self, event_type: str, observer: BaseAbstractObserver) -> None:
self._subscriptions[event_type].append(observer)
def unsubscribe(self, event_type: str, observer: BaseAbstractObserver) -> None:
self._subscriptions[event_type].remove(observer)
def notify(self, event: BaseEvent) -> None:
for observer in self._subscriptions.get(event.event_type, []):
observer.update(event)
Additional Information
• Base Event Class: All events should inherit from django_event_observer.events.base.BaseEvent.
• EventManager Methods:
• subscribe(event_type: str, observer: BaseAbstractObserver): Subscribes an observer to an event type.
• unsubscribe(event_type: str, observer: BaseAbstractObserver): Unsubscribes an observer from an event type.
• notify(event: BaseEvent): Notifies all observers subscribed to event.event_type.
Raw data
{
"_id": null,
"home_page": null,
"name": "django-event-observer",
"maintainer": null,
"docs_url": null,
"requires_python": "~=3.13",
"maintainer_email": null,
"keywords": null,
"author": null,
"author_email": "\"Sergei (Gefest) Romanchuk\" <pod.cargoes.0u@icloud.com>",
"download_url": "https://files.pythonhosted.org/packages/75/ba/1501214506992547634fb348df2e02d1c54875dc961ef142586e71ceb49b/django_event_observer-2024.1.1.tar.gz",
"platform": null,
"description": "# django_event_observer\n\n\ndjango_event_observer is a fully-fledged Django application designed for observing and handling events within your project. It provides a flexible and extensible framework for subscribing to various event types and reacting to them accordingly.\n\nInstallation\n\nInstall the package using pip:\n\npip install django_event_observer\n\nConfiguration\n\nAfter installation, add django_event_observer to the INSTALLED_APPS in your Django settings file:\n\nINSTALLED_APPS = [\n # ...\n 'django_event_observer',\n # ...\n]\n\nDefault Settings\n\nBy default, the DJANGO_EVENT_OBSERVER settings are as follows:\n\nfrom types import MappingProxyType\n\nDJANGO_EVENT_OBSERVER = MappingProxyType(\n {\n \"EVENT_MANAGER\": \"django_event_observer.event_manager.EventManager\",\n \"AUTO_OBSERVERS\": [],\n }\n)\n\nYou can override the DJANGO_EVENT_OBSERVER variable in your project\u2019s settings to specify a custom import path for the EventManager or to add automatic observers.\n\nCustomizing EventManager\n\nIf you need to use a custom EventManager, specify the import path in your settings:\n\nDJANGO_EVENT_OBSERVER = {\n \"EVENT_MANAGER\": \"my_app.event_manager.CustomEventManager\",\n \"AUTO_OBSERVERS\": [],\n}\n\nAutomatic Observers\n\nThe AUTO_OBSERVERS parameter is a list of dictionaries, each describing an observer. These observers are classes inheriting from django_event_observer.observers.abstract.base.BaseAbstractObserver.\n\nSyntax:\n```python\n\n\nDJANGO_EVENT_OBSERVER = {\n \"EVENT_MANAGER\": \"django_event_observer.event_manager.EventManager\",\n \"AUTO_OBSERVERS\": [\n {\n \"CLASS\": \"my_module.observers.MyObserver\",\n \"INIT_ARGS\": [\"arg1\", \"arg2\"],\n \"INIT_KWARGS\": {\"key1\": \"value1\", \"key2\": \"value2\"},\n \"EVENT_TYPES\": [\"event_type_1\", \"event_type_2\"],\n },\n # Other observers...\n ],\n}\n```\n\n\t\u2022\t\"CLASS\": A string with the import path to the observer class.\n\t\u2022\t\"INIT_ARGS\": A list of positional arguments for class initialization.\n\t\u2022\t\"INIT_KWARGS\": A dictionary of keyword arguments for class initialization.\n\t\u2022\t\"EVENT_TYPES\": A list of strings representing event types that the observer should subscribe to.\n\n##Usage\n\nCreating a Custom Observer\n\nTo create a custom observer, inherit from BaseAbstractObserver and implement the update method:\n\n# my_module/observers.py\n```python\n\n\nfrom django_event_observer.observers.abstract.base import BaseAbstractObserver\nfrom django_event_observer.events.base import BaseEvent\n\nclass MyObserver(BaseAbstractObserver):\n def __init__(self, *args, **kwargs):\n super().__init__()\n # Additional initialization...\n\n def update(self, event: BaseEvent) -> None:\n # Logic to handle the event\n print(f\"Received event: {event}\")\n```\n\nDefinition of BaseAbstractObserver:\n\n# django_event_observer/observers/abstract/base.py\n\nfrom abc import ABC, abstractmethod\nfrom ...events.base import BaseEvent\n\nclass BaseAbstractObserver(ABC):\n @abstractmethod\n def update(self, event: BaseEvent) -> None:\n ...\n\nAny custom observer must inherit from BaseAbstractObserver and implement the update method, which will be called when an event occurs.\n\nCreating Custom Events\n\nEvents should inherit from BaseEvent and can use generics for typing event_data:\n\n# my_module/events.py\n\nfrom django_event_observer.events.base import BaseEvent\nfrom typing import Generic, TypeVar\n\nT = TypeVar('T')\n\nclass MyEvent(BaseEvent[T], Generic[T]):\n def __init__(self, event_type: str, event_data: T):\n super().__init__(event_type, event_data)\n # Additional initialization...\n\n def __str__(self):\n return f\"{self.event_type}: {self.event_data}\"\n\nDefinition of BaseEvent:\n\n# django_event_observer/events/base.py\n\nimport typing\nfrom django.apps import apps\n\nif typing.TYPE_CHECKING:\n from ..event_manager import EventManager\n\n_EventDataT = typing.TypeVar(\"_EventDataT\")\n\nclass BaseEvent(typing.Generic[_EventDataT]):\n def __init__(self, event_type: str, event_data: _EventDataT):\n self._event_manager = apps.get_app_config(\"django_event_observer\").event_manager\n self.event_type = event_type\n self.event_data = event_data\n\n def __call__(self):\n self.notify()\n\n def notify(self):\n self._event_manager.notify(self)\n\n def __str__(self):\n return f\"{self.event_type}: {self.event_data}\"\n\nAccessing EventManager\n\nTo subscribe or unsubscribe an observer from specific event types, you need to access the initialized EventManager. It is recommended to do this using the Django applications registry:\n\nfrom django.apps import apps\n\nevent_manager = apps.get_app_config(\"django_event_observer\").event_manager\n\nSubscribing to Events\n\nTo subscribe an observer to a specific event type:\n\nfrom my_module.observers import MyObserver\nfrom django.apps import apps\n\n# Get the EventManager instance\nevent_manager = apps.get_app_config(\"django_event_observer\").event_manager\n\n# Create an observer instance\nobserver = MyObserver()\n\n# Subscribe the observer to an event type\nevent_manager.subscribe(\"my_event_type\", observer)\n\nUnsubscribing from Events\n\nTo unsubscribe an observer from a specific event type:\n\n# Unsubscribe the observer from the event type\nevent_manager.unsubscribe(\"my_event_type\", observer)\n\nSending Events\n\nTo send an event and notify all subscribed observers, create an event instance and call the notify method:\n\nfrom my_module.events import MyEvent\n\n# Create an event instance\nevent = MyEvent(event_type=\"my_event_type\", event_data={\"key\": \"value\"})\n\n# Notify observers\nevent.notify()\n\nOr you can call the event instance as a function thanks to the __call__ method:\n\n# Notify observers by calling the event\nevent()\n\nYou can also use the EventManager directly to notify:\n\n# Notify observers via EventManager\nevent_manager.notify(event)\n\nFull Usage Example\n\n# observers.py\n\nfrom django_event_observer.observers.abstract.base import BaseAbstractObserver\nfrom django_event_observer.events.base import BaseEvent\n\nclass MyObserver(BaseAbstractObserver):\n def update(self, event: BaseEvent) -> None:\n print(f\"Observer received event: {event}\")\n\n# events.py\n\nfrom django_event_observer.events.base import BaseEvent\n\nclass UserRegisteredEvent(BaseEvent[dict]):\n def __init__(self, user_data: dict):\n super().__init__(event_type=\"user_registered\", event_data=user_data)\n\n def __str__(self):\n return f\"UserRegisteredEvent: {self.event_data}\"\n\n# main.py\n\nfrom django.apps import apps\nfrom my_module.observers import MyObserver\nfrom my_module.events import UserRegisteredEvent\n\n# Get the EventManager instance\nevent_manager = apps.get_app_config(\"django_event_observer\").event_manager\n\n# Create an observer instance\nobserver = MyObserver()\n\n# Subscribe the observer to the \"user_registered\" event\nevent_manager.subscribe(\"user_registered\", observer)\n\n# Create a user registration event\nuser_data = {\"username\": \"john_doe\", \"email\": \"john@example.com\"}\nevent = UserRegisteredEvent(user_data)\n\n# Notify observers\nevent.notify()\n\n# Or call the event as a function\nevent()\n\n# Unsubscribe the observer when no longer needed\nevent_manager.unsubscribe(\"user_registered\", observer)\n\nInternal Structure of EventManager\n\nThe standard EventManager looks like this:\n\n# django_event_observer/event_manager.py\n\nfrom collections import defaultdict\nimport typing\nfrom .observers.abstract.base import BaseAbstractObserver\nfrom .events.base import BaseEvent\n\n_ObserversT = dict[str, list[BaseAbstractObserver]]\n\nclass EventManager:\n _subscriptions: typing.ClassVar[_ObserversT] = defaultdict(list)\n\n def subscribe(self, event_type: str, observer: BaseAbstractObserver) -> None:\n self._subscriptions[event_type].append(observer)\n\n def unsubscribe(self, event_type: str, observer: BaseAbstractObserver) -> None:\n self._subscriptions[event_type].remove(observer)\n\n def notify(self, event: BaseEvent) -> None:\n for observer in self._subscriptions.get(event.event_type, []):\n observer.update(event)\n\nAdditional Information\n\n\t\u2022\tBase Event Class: All events should inherit from django_event_observer.events.base.BaseEvent.\n\t\u2022\tEventManager Methods:\n\t\u2022\tsubscribe(event_type: str, observer: BaseAbstractObserver): Subscribes an observer to an event type.\n\t\u2022\tunsubscribe(event_type: str, observer: BaseAbstractObserver): Unsubscribes an observer from an event type.\n\t\u2022\tnotify(event: BaseEvent): Notifies all observers subscribed to event.event_type.\n",
"bugtrack_url": null,
"license": "MIT License Copyright (c) 2024 Sergey Romanchuk Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the \"Software\"), to deal in the Software without restriction, including without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to permit persons to whom the Software is furnished to do so, subject to the following conditions: The above copyright notice and this permission notice shall be included in all copies or substantial portions of the Software. THE SOFTWARE IS PROVIDED \"AS IS\", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. ",
"summary": null,
"version": "2024.1.1",
"project_urls": {
"homepage": "https://github.com/GefMar/django_event_observer"
},
"split_keywords": [],
"urls": [
{
"comment_text": "",
"digests": {
"blake2b_256": "d3cf2bd2255b9e787a23d3c5d3f835b913ba670fafac9fa0dc6a54016afb2767",
"md5": "79eb3ac6f23bbf008eed5e90f35f1ff2",
"sha256": "e0e9b4bc4d360a86ef6cebf9261a6215532d6ae8faee3cda6b593de3d9610de4"
},
"downloads": -1,
"filename": "django_event_observer-2024.1.1-py3-none-any.whl",
"has_sig": false,
"md5_digest": "79eb3ac6f23bbf008eed5e90f35f1ff2",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": "~=3.13",
"size": 9886,
"upload_time": "2024-10-21T18:56:20",
"upload_time_iso_8601": "2024-10-21T18:56:20.570468Z",
"url": "https://files.pythonhosted.org/packages/d3/cf/2bd2255b9e787a23d3c5d3f835b913ba670fafac9fa0dc6a54016afb2767/django_event_observer-2024.1.1-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": "",
"digests": {
"blake2b_256": "75ba1501214506992547634fb348df2e02d1c54875dc961ef142586e71ceb49b",
"md5": "66b0cc960c9e09abfb4630456830e17b",
"sha256": "3333747fe16650e545763cb2ea7302ca53abd6eda6fb002b65150bcc07a14367"
},
"downloads": -1,
"filename": "django_event_observer-2024.1.1.tar.gz",
"has_sig": false,
"md5_digest": "66b0cc960c9e09abfb4630456830e17b",
"packagetype": "sdist",
"python_version": "source",
"requires_python": "~=3.13",
"size": 9054,
"upload_time": "2024-10-21T18:56:21",
"upload_time_iso_8601": "2024-10-21T18:56:21.685873Z",
"url": "https://files.pythonhosted.org/packages/75/ba/1501214506992547634fb348df2e02d1c54875dc961ef142586e71ceb49b/django_event_observer-2024.1.1.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2024-10-21 18:56:21",
"github": true,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"github_user": "GefMar",
"github_project": "django_event_observer",
"travis_ci": false,
"coveralls": false,
"github_actions": true,
"lcname": "django-event-observer"
}
