A Django app that tracks software subscriptions and fires reminders before they renew. It works out when to call your notification adapter and what to pass it. You write the adapter.
uv add sub-amigo
# or
pip install sub-amigoAdd to INSTALLED_APPS and run migrations:
INSTALLED_APPS = [
...
"sub_amigo",
]python manage.py migrateSubclass BaseNotificationAdapter and implement send(). Return True on success, False on failure. Do not raise exceptions.
# myapp/adapters.py
import logging
from django.core.mail import send_mail
from django.conf import settings
from sub_amigo import BaseNotificationAdapter, NotificationPayload
logger = logging.getLogger(__name__)
class EmailAdapter(BaseNotificationAdapter):
def send(self, payload: NotificationPayload) -> bool:
try:
send_mail(
subject=payload.subject,
message=payload.message,
from_email=settings.DEFAULT_FROM_EMAIL,
recipient_list=[payload.recipient],
)
return True
except Exception:
logger.exception("Reminder failed for %s", payload.recipient)
return FalseNotificationPayload has four fields:
| Field | Description |
|---|---|
recipient |
The value from Subscription.recipient_ref, passed through unchanged |
subject |
Pre-rendered subject line |
message |
Fully-rendered body text |
metadata |
Dict with subscription_id, billing_date, days_before, cost, currency |
# settings.py
SUB_AMIGO_ADAPTER = "myapp.adapters.EmailAdapter"Run once a day via cron or a task scheduler:
python manage.py process_remindersTo process a specific date instead of today:
python manage.py process_reminders --date 2024-08-01With Celery beat:
# celery.py
app.conf.beat_schedule = {
"process-reminders": {
"task": "myapp.tasks.process_reminders",
"schedule": crontab(hour=8, minute=0),
},
}# myapp/tasks.py
from celery import shared_task
from sub_amigo.engine import SubscriptionReminderEngine
from myapp.adapters import EmailAdapter
@shared_task
def process_reminders():
SubscriptionReminderEngine(adapter=EmailAdapter()).process_daily_reminders()process_reminders() and process_daily_reminders() both return a ProcessingResult:
result = amigo.process_reminders()
print(result.succeeded) # reminders sent
print(result.failed) # adapter returned False or template error
print(result.skipped) # already sent for this billing cycle
print(result.processed) # total rules evaluated| Field | Type | Description |
|---|---|---|
run_date |
date |
The date the engine ran against |
processed |
int |
Total rules evaluated |
succeeded |
int |
Reminders sent successfully |
failed |
int |
Adapter failures or template errors |
skipped |
int |
Already sent for this billing cycle |
results |
list[ReminderResult] |
Per-rule detail |
Each ReminderResult has rule_id, subscription_id, subscription_name, success, skipped, and error.
The SubAmigo class is the quickest way to get started. It creates the subscription and its reminder rules in one call and exposes process_reminders() on the same object.
from datetime import date
from decimal import Decimal
from sub_amigo.service import SubAmigo
from myapp.adapters import EmailAdapter
amigo = SubAmigo(adapter=EmailAdapter())
sub = amigo.subscribe(
name="GitHub Teams",
cost=Decimal("4.00"),
currency="USD",
next_billing_date=date(2024, 8, 1),
recipient_ref="billing@example.com",
owner_ref="team-42",
remind_days_before=[7, 3, 0], # uses the default template
)
# Run reminders (call this daily)
result = amigo.process_reminders()For custom message templates per rule, use reminder_specs instead:
from sub_amigo.service import ReminderSpec
sub = amigo.subscribe(
name="GitHub Teams",
cost=Decimal("4.00"),
next_billing_date=date(2024, 8, 1),
recipient_ref="billing@example.com",
reminder_specs=[
ReminderSpec(days_before=7, message_template="One week until {subscription_name} renews."),
ReminderSpec(days_before=0, message_template="{subscription_name} charges today ({cost} {currency})."),
],
)Both arguments can be mixed. remind_days_before uses the default template; reminder_specs uses whatever you provide.
| Variable | Value |
|---|---|
{subscription_name} |
Name of the subscription |
{cost} |
Billing amount |
{currency} |
ISO 4217 currency code |
{next_billing_date} |
Next billing date, YYYY-MM-DD |
{days_before} |
Days until billing |
{description} |
Subscription description, may be empty |
After a subscription charges, move it to the next cycle:
sub.advance_billing_date().save()| Interval | Behaviour |
|---|---|
monthly |
One calendar month forward, month-end dates handled correctly |
annually |
One calendar year forward |
custom |
Forward by billing_interval_days days |
sub-amigo fires Django signals after every reminder attempt so your app can react without subclassing anything.
from django.dispatch import receiver
from sub_amigo.signals import reminder_sent, reminder_failed
@receiver(reminder_sent)
def log_to_datadog(sender, subscription, rule, payload, log, **kwargs):
datadog.increment("reminders.sent", tags=[f"sub:{subscription.name}"])
@receiver(reminder_failed)
def alert_on_call(sender, subscription, rule, error, log, **kwargs):
pagerduty.trigger(f"{subscription.name} reminder failed: {error}")reminder_sent kwargs:
| kwarg | Type |
|---|---|
subscription |
Subscription |
rule |
ReminderRule |
payload |
NotificationPayload |
log |
NotificationLog |
reminder_failed kwargs:
| kwarg | Type |
|---|---|
subscription |
Subscription |
rule |
ReminderRule |
error |
str |
log |
NotificationLog |
reminder_failed fires on both adapter failures and template render errors.
NotificationLog records every fired reminder keyed on (reminder_rule, billing_cycle_date). That pair has a unique constraint, so running process_daily_reminders() twice on the same date does not send duplicates. The engine checks before calling the adapter; the database constraint enforces it regardless.
| Field | Type | Notes |
|---|---|---|
id |
UUID | Auto-generated |
name |
CharField | |
cost |
DecimalField | |
currency |
CharField | ISO 4217, default "USD" |
billing_interval |
CharField | monthly / annually / custom |
billing_interval_days |
PositiveIntegerField | Required when billing_interval = "custom", min 1 |
next_billing_date |
DateField | |
status |
CharField | active / paused / cancelled |
recipient_ref |
CharField | Passed to your adapter unchanged |
owner_ref |
CharField | Optional, for multi-tenant filtering |
metadata |
JSONField | Arbitrary app-specific data |
| Field | Type | Notes |
|---|---|---|
subscription |
ForeignKey | Cascade delete |
days_before |
PositiveSmallIntegerField | 0 fires on the billing date; unique per subscription |
message_template |
TextField | str.format_map() template |
is_active |
BooleanField |
Read-only ledger. One row per reminder fired.
For data migrations, bulk imports, management commands, or admin actions, you can work with the ORM directly.
from sub_amigo.models import Subscription, ReminderRule, BillingInterval
sub = Subscription.objects.create(
name="GitHub Teams",
cost="4.00",
currency="USD",
billing_interval=BillingInterval.MONTHLY,
next_billing_date=date(2024, 8, 1),
recipient_ref="billing@example.com",
)
ReminderRule.objects.create(
subscription=sub,
days_before=7,
message_template="{subscription_name} renews on {next_billing_date}.",
)SubAmigo.subscribe() does the same thing internally. Reach for the ORM when you need finer control or are operating outside a request cycle.
- Python 3.12+
- Django 5.0+
- python-dateutil 2.9+
MIT