FastAPI Event
Project description
FastAPI event
How to use
- Inject the
event_emitter
into Service dependencies to use
# user_service.py
from fastapi_event import get_event_emitter
class UserService:
def __init__(self, event_emitter=get_event_emitter()):
self.event_emitter = event_emitter
async def deactivate_user(self, user: UserEntity):
await self.user_repo.update_by_id(user.id, data={'deactivated_at': datetime.datetime.utcnow()})
await self.event_emitter.emit(UserEvent.DEACTIVATED, user.id)
- Create
observer
file to listenevent
and handle it
# users/observers/user_observer.py
from fastapi_event import observer, event
@observer('user:*')
class UserObserver:
def __init__(self, notification_service: NotificationService = Depends()):
self.notification_service = notification_service
@event(UserEvent.DEACTIVATED)
async def handle_user_deactivated_for_related_user_tasks(self, user_id: UUID):
# Add logic here
If we need to update user related jobs in another module, we should create the UserObserver
in that module followed by modular based
too
Such as, we need to update caches of user posts
# posts/observers/user_observer.py
from fastapi_event import observer, event
@observer('user:*')
class UserObserver:
@event(UserEvent.DEACTIVATED)
async def handle_user_deactivated_for_related_post_tasks(self, user_id: UUID):
await queue.add_to_queue(PostTask.CACHE_COUNT_COMMENT, user_id)
Caveat
Currently, the @observer has a limitation that when I inject a service and that service's dependencies are injected with an event_emitter, it will encounter an erjor.
Right now, to fix this issue we should move the affected service to a separate one without injecting the event_emitter, or directly importing the repository to execute the queries.
It happens because we need to collect all the observer
class before injecting the event_emitter
dependency inside that service.
Example:
Let's say we have a feature called deactivated user
. We will emit the deactivate_user
event then handle it in the observer
Inside the observer
after hanlding what needes to be done, we will push notification to user to force them logout
To do that, we want to inject the NotificationService
, inside the NotificationService
dependencies we have some other tasks
that need to inject the MediaService
.
The issue happens here. The MediaService
dependencies we inject the event_emitter
that will lead to the rest observers
will never collect any __Observer
class anymore.
# media_service.py
from fastapi_event import get_event_emitter
class MediaService:
def __init__(self, event_emitter=get_event_emitter()),
):
self.media_service = media_service
async def create(self):
# Logic
# notification_service.py
class NotificationService:
def __init__(self, media_service: MediaService = Depends()):
self.media_service = media_service
async def emit(self):
# Logic
# users/observers/user_observer.py
from fastapi_event import observer, event
@observer('user:*')
class UserObserver:
def __init__(self, notification_service: NotificationService = Depends()):
self.notification_service = notification_service
@event(UserEvent.DEACTIVATED)
async def handle_user_deactivated_for_related_user_tasks(self, user_id: UUID):
# Add logic here
To temporarily resolve this issue, we have 2 ways:
- Try to not inject
MediaService
inside theNotificationService
- Move the
NotificationService
method inside theUserObserver
and inject thenotification_repo
directly
How to test in testpypi
- Increase the version in
pyproject.toml
- Run command
$ . ./build_and_test.sh
How to publish new version
- Increase the version in
pyproject.toml
- Run command
$ . ./build_and_publish.sh
Project details
Release history Release notifications | RSS feed
Download files
Download the file for your platform. If you're not sure which to choose, learn more about installing packages.
Source Distribution
Built Distribution
Hashes for fastapi_event_emitter-0.0.2.tar.gz
Algorithm | Hash digest | |
---|---|---|
SHA256 | b65ef0088b3d1401da8ed3c202d41a70279201f81910abfe3dc50a58e6fb2be4 |
|
MD5 | f21e851ca5c35eed028ae2d721b68f9d |
|
BLAKE2b-256 | 2eae2f51b8705ce4f34d464a72992d5518d15ccf04bad91d578d01137217869d |
Hashes for fastapi_event_emitter-0.0.2-py3-none-any.whl
Algorithm | Hash digest | |
---|---|---|
SHA256 | 679124a523c5e217c9fbfa159e3d160760b5f6f19918da9ba7790b00db079eb1 |
|
MD5 | 0223e161434e1a08de24ef0fd8eba734 |
|
BLAKE2b-256 | 99d5f12dc6e4a0730ff669f263cf5da305d3b219ee656c4e61d58d0a71b10622 |