Python Novu SDK
The Python Novu SDK and package provides a fluent and expressive interface for interacting with Novu's API and managing notifications.
Install
To install this package
# Via pip
pip install novu
# Via poetry
poetry add novu
Contents
Quick start
This package is a wrapper of all the resources offered by Novu, we will just start by triggering an event on Novu.
To do this, you will need to:
- Create your first notification workflow and keep in mind the identifier to trigger the workflow: https://docs.novu.co/overview/quickstart/general-quickstart#create-a-workflow
- Retrieve your API key from the Novu dashboard directly in the settings section: https://web.novu.co/settings
- Write code to trigger your first event:
from novu.api import EventApi
event_api = EventApi("https://api.novu.co", "<NOVU_API_KEY>")
event_api.trigger(
name="<YOUR_WORKFLOW_ID>",
recipients="<YOUR_SUBSCRIBER_ID>",
payload={},
)
This will trigger a notification to the subscribers.
Code Snippet Examples
Events
Firstly, make imports and declare the needed variables this way:
from novu.api import EventApi
url = "https://api.novu.co"
api_key = "<NOVU_API_KEY>"
Trigger an event - Send notification to subscribers:
from novu.api import EventApi
novu = EventApi(url, api_key).trigger(
name="digest-workflow-example",
recipients="<SUBSCRIBER_IDENTIFIER>",
payload={},
)
Bulk Trigger events - Trigger multiple events at once:
from novu.dto.event import InputEventDto
from novu.api import EventApi
url = "https://api.novu.co"
api_key = "<NOVU_API_KEY>"
event_1 = InputEventDto(
name="digest-workflow-example",
recipients="<SUBSCRIBER_IDENTIFIER>",
payload={},
)
event_2 = InputEventDto(
name="digest-workflow-example",
recipients="<SUBSCRIBER_IDENTIFIER>",
payload={},
)
novu = EventApi("https://api.novu.co", api_key).trigger_bulk(events=[event1, event2])
Include actor field:
from novu.api import EventApi
novu = EventApi(url, api_key).trigger(
name="workflow_trigger_identifier",
recipients="subscriber_id",
actor={
"subscriberId": "subscriber_id_actor"
},
payload={
"key":"value"
},
)
Broadcast to all current subscribers:
novu = EventApi(url, api_key).broadcast(
name="digest-workflow-example",
payload={"customVariable": "value"},
)
Subscribers
from novu.dto.subscriber import SubscriberDto
from novu.api.subscriber import SubscriberApi
url = "https://api.novu.co"
api_key = "<NOVU_API_KEY>"
subscriber = SubscriberDto(
email="novu_user@mail.com",
subscriber_id="82a48af6ac82b3cc2157b57f",
first_name="",
last_name="",
phone="",
avatar="",
)
novu = SubscriberApi(url, api_key).create(subscriber)
novu = SubscriberApi(url, api_key).get(subscriber_id)
novu = SubscriberApi(url, api_key).list()
Topics
from novu.api import TopicApi
url = "<NOVU_URL>"
api_key = "<NOVU_API_KEY>"
novu = TopicApi(url, api_key).create(
key="new-customers", name="New business customers"
)
novu = TopicApi(url, api_key).get(key="new-customers")
novu = TopicApi(url, api_key).list()
novu = TopicApi(url, api_key).rename(key="new-customers", name="New business customers")
novu = TopicApi(url, api_key).subscribe(key="old-customers", subscribers="<LIST_OF_SUBSCRIBER_IDs>")
novu = TopicApi(url, api_key).unsubscribe(key="old-customers", subscribers="<LIST_OF_SUBSCRIBER_IDs>")
Feeds
from novu.api.feed import FeedApi
url = "<NOVU_URL>"
api_key = "<NOVU_API_KEY>"
novu = FeedApi(url, api_key).create(name="<SUPPLY_NAME_FOR_FEED>")
FeedApi(url, api_key).delete(feed_id="<FEED_NOVU_INTERNAL_ID>")
novu = FeedApi(url, api_key).list()
Environments
from novu.api.environment import EnvironmentApi
url = "<NOVU_URL>"
api_key = "<NOVU_API_KEY>"
novu = EnvironmentApi(url, api_key).create(
name="<INSERT_NAME>",
parent_id="<INSERT_PARENT_ID>"
)
novu = EnvironmentApi(url, api_key).list()
novu = EnvironmentApi(url, api_key).current()
novu = EnvironmentApi(url, api_key).api_keys()
Tenants
from novu.api.tenant import TenantApi
url = "<NOVU_URL>"
api_key = "<NOVU_API_KEY>"
tenant = TenantApi(url, api_key).create(
identifier="<INSERT_UNIQUE_TENANT_ID>",
name="<INSERT_NAME>",
data={}
)
tenants = TenantApi(url, api_key).list()
tenants = TenantApi(url, api_key).list(page=1, limit=10)
tenant = TenantApi(url, api_key).get("<TENANT-IDENTIFIER>")
tenant = TenantApi(url, api_key).patch(
"<CURRENT-TENANT-IDENTIFIER>",
identifier="<NEW-IDENTIFIER>",
name="<NEW-NAME>",
data="<NEW-DATA>"
)
TenantApi(url, api_key).delete("<TENANT-IDENTIFIER>")
Go further
After a quick start with the SDK, you'll quickly get to grips with the advanced use of the SDK and the other APIs available.
For this purpose, documentation is available here: https://novu-python.readthedocs.io/
Development
poetry install
poetry run pre-commit install --install-hook
poetry run pre-commit install --install-hooks --hook-type commit-msg
Contributing
Feature requests, bug reports and pull requests are welcome. Please create an issue.
Support and Feedback
Be sure to visit the Novu official documentation website for additional information about our SDK.
If you need additional assistance, join our Discord server here.
License
Novu Python SDK is licensed under the MIT License - see the LICENSE file for details.