Skip to content

Commit 16fb53c

Browse files
alexeygrigorevalexeygrigorev
committed
Add recipient list import jobs
1 parent a0d528b commit 16fb53c

8 files changed

Lines changed: 698 additions & 2 deletions

File tree

‎mailing/management/commands/process_recipient_list_imports.py‎

Lines changed: 21 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,21 @@
1+
from django.core.management.base import BaseCommand
2+
3+
from mailing.services.recipient_lists import process_pending_recipient_list_import_jobs
4+
5+
6+
class Command(BaseCommand):
7+
help = "Process pending recipient-list import jobs."
8+
9+
def add_arguments(self, parser):
10+
parser.add_argument(
11+
"--limit",
12+
type=int,
13+
default=25,
14+
help="Maximum number of pending import jobs to process.",
15+
)
16+
17+
def handle(self, *args, **options):
18+
result = process_pending_recipient_list_import_jobs(limit=options["limit"])
19+
self.stdout.write(
20+
"processed={processed} succeeded={succeeded} failed={failed}".format(**result)
21+
)

‎mailing/migrations/0022_recipientlistimportjob.py‎

Lines changed: 47 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,47 @@
1+
# Generated by Django 6.0.5 on 2026-06-26 18:53
2+
3+
import django.db.models.deletion
4+
from django.db import migrations, models
5+
6+
7+
class Migration(migrations.Migration):
8+
9+
dependencies = [
10+
('mailing', '0021_campaign_recipient_list_key'),
11+
]
12+
13+
operations = [
14+
migrations.CreateModel(
15+
name='RecipientListImportJob',
16+
fields=[
17+
('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
18+
('created_at', models.DateTimeField(auto_now_add=True)),
19+
('updated_at', models.DateTimeField(auto_now=True)),
20+
('list_key', models.CharField(max_length=255)),
21+
('source_url', models.URLField(max_length=2048)),
22+
('idempotency_key', models.CharField(blank=True, max_length=255)),
23+
('status', models.CharField(choices=[('pending', 'Pending'), ('processing', 'Processing'), ('succeeded', 'Succeeded'), ('failed', 'Failed')], db_index=True, default='pending', max_length=20)),
24+
('list_defaults', models.JSONField(blank=True, default=dict)),
25+
('remove_absent', models.BooleanField(default=False)),
26+
('row_count', models.PositiveIntegerField(default=0)),
27+
('created_count', models.PositiveIntegerField(default=0)),
28+
('updated_count', models.PositiveIntegerField(default=0)),
29+
('removed_count', models.PositiveIntegerField(default=0)),
30+
('failed_count', models.PositiveIntegerField(default=0)),
31+
('failed_rows', models.JSONField(blank=True, default=list)),
32+
('content_sha256', models.CharField(blank=True, max_length=64)),
33+
('error', models.TextField(blank=True)),
34+
('started_at', models.DateTimeField(blank=True, null=True)),
35+
('completed_at', models.DateTimeField(blank=True, null=True)),
36+
('audience', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='recipient_list_import_jobs', to='mailing.audience')),
37+
('client', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='recipient_list_import_jobs', to='mailing.client')),
38+
('recipient_list', models.ForeignKey(blank=True, null=True, on_delete=django.db.models.deletion.SET_NULL, related_name='import_jobs', to='mailing.recipientlist')),
39+
],
40+
options={
41+
'db_table': 'recipient_list_import_jobs',
42+
'ordering': ['-created_at', '-id'],
43+
'indexes': [models.Index(fields=['client', 'audience', 'status'], name='recip_import_scope_status_idx'), models.Index(fields=['list_key', 'created_at'], name='recip_import_list_created_idx')],
44+
'constraints': [models.UniqueConstraint(condition=models.Q(('idempotency_key', ''), _negated=True), fields=('client', 'idempotency_key'), name='unique_recipient_list_import_job_idempotency')],
45+
},
46+
),
47+
]

‎mailing/models.py‎

Lines changed: 58 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -350,6 +350,64 @@ def __str__(self):
350350
return f"{self.recipient_list.key}: {self.source_object_key}"
351351

352352

353+
class RecipientListImportJobStatus(models.TextChoices):
354+
PENDING = "pending", "Pending"
355+
PROCESSING = "processing", "Processing"
356+
SUCCEEDED = "succeeded", "Succeeded"
357+
FAILED = "failed", "Failed"
358+
359+
360+
class RecipientListImportJob(TimeStampedModel):
361+
client = models.ForeignKey(Client, on_delete=models.CASCADE, related_name="recipient_list_import_jobs")
362+
audience = models.ForeignKey(Audience, on_delete=models.CASCADE, related_name="recipient_list_import_jobs")
363+
recipient_list = models.ForeignKey(
364+
RecipientList,
365+
on_delete=models.SET_NULL,
366+
related_name="import_jobs",
367+
null=True,
368+
blank=True,
369+
)
370+
list_key = models.CharField(max_length=255)
371+
source_url = models.URLField(max_length=2048)
372+
idempotency_key = models.CharField(max_length=255, blank=True)
373+
status = models.CharField(
374+
max_length=20,
375+
choices=RecipientListImportJobStatus.choices,
376+
default=RecipientListImportJobStatus.PENDING,
377+
db_index=True,
378+
)
379+
list_defaults = models.JSONField(default=dict, blank=True)
380+
remove_absent = models.BooleanField(default=False)
381+
row_count = models.PositiveIntegerField(default=0)
382+
created_count = models.PositiveIntegerField(default=0)
383+
updated_count = models.PositiveIntegerField(default=0)
384+
removed_count = models.PositiveIntegerField(default=0)
385+
failed_count = models.PositiveIntegerField(default=0)
386+
failed_rows = models.JSONField(default=list, blank=True)
387+
content_sha256 = models.CharField(max_length=64, blank=True)
388+
error = models.TextField(blank=True)
389+
started_at = models.DateTimeField(null=True, blank=True)
390+
completed_at = models.DateTimeField(null=True, blank=True)
391+
392+
class Meta:
393+
db_table = "recipient_list_import_jobs"
394+
ordering = ["-created_at", "-id"]
395+
constraints = [
396+
models.UniqueConstraint(
397+
fields=["client", "idempotency_key"],
398+
condition=~models.Q(idempotency_key=""),
399+
name="unique_recipient_list_import_job_idempotency",
400+
),
401+
]
402+
indexes = [
403+
models.Index(fields=["client", "audience", "status"], name="recip_import_scope_status_idx"),
404+
models.Index(fields=["list_key", "created_at"], name="recip_import_list_created_idx"),
405+
]
406+
407+
def __str__(self):
408+
return f"{self.client.slug}/{self.audience.slug}/{self.list_key} import {self.status}"
409+
410+
353411
class CampaignStatus(models.TextChoices):
354412
DRAFT = "draft", "Draft"
355413
QUEUED = "queued", "Queued"

‎mailing/services/api_docs.py‎

Lines changed: 116 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -4,7 +4,13 @@
44
from django.conf import settings
55
from django.urls import reverse
66

7-
from mailing.models import CampaignStatus, EmailValidationStatus, RecipientListType, SubscriptionStatus
7+
from mailing.models import (
8+
CampaignStatus,
9+
EmailValidationStatus,
10+
RecipientListImportJobStatus,
11+
RecipientListType,
12+
SubscriptionStatus,
13+
)
814

915
DEMO_API_KEYS = [
1016
{
@@ -59,6 +65,8 @@
5965
"mailing:api_recipient_list_member": "/api/recipient-lists/{list_key}/members/{source_object_key}",
6066
"mailing:api_recipient_list_bulk_upsert": "/api/recipient-lists/{list_key}/members/bulk-upsert",
6167
"mailing:api_recipient_list_reconcile": "/api/recipient-lists/{list_key}/members/reconcile",
68+
"mailing:api_recipient_list_imports": "/api/recipient-lists/{list_key}/imports",
69+
"mailing:api_recipient_list_import": "/api/recipient-lists/{list_key}/imports/{job_id}",
6270
"mailing:api_recipient_list_transactional_send": "/api/recipient-lists/{list_key}/transactional-send",
6371
"mailing:api_transient_recipient_list_transactional_send": (
6472
"/api/transient-recipient-lists/transactional-send"
@@ -804,6 +812,16 @@ def endpoint_groups():
804812
"/api/recipient-lists/{list_key}/members/reconcile",
805813
"Replace list membership from a current source snapshot.",
806814
),
815+
(
816+
"POST",
817+
"/api/recipient-lists/{list_key}/imports",
818+
"Create a pending JSONL import job from a fetchable URL.",
819+
),
820+
(
821+
"GET",
822+
"/api/recipient-lists/{list_key}/imports/{job_id}",
823+
"Get recipient-list import job status.",
824+
),
807825
(
808826
"POST",
809827
"/api/recipient-lists/{list_key}/transactional-send",
@@ -889,6 +907,14 @@ def route_path_map():
889907
"mailing:api_recipient_list_reconcile",
890908
args=["ml-zoomcamp-2026"],
891909
),
910+
API_DOC_PATHS["mailing:api_recipient_list_imports"]: reverse(
911+
"mailing:api_recipient_list_imports",
912+
args=["ml-zoomcamp-2026"],
913+
),
914+
API_DOC_PATHS["mailing:api_recipient_list_import"]: reverse(
915+
"mailing:api_recipient_list_import",
916+
args=["ml-zoomcamp-2026", 101],
917+
),
892918
API_DOC_PATHS["mailing:api_recipient_list_transactional_send"]: reverse(
893919
"mailing:api_recipient_list_transactional_send",
894920
args=["ml-zoomcamp-2026:@e:@homework:homework-1"],
@@ -955,6 +981,7 @@ def bearer_responses(success, *, accepted=False):
955981
MESSAGE_ID_PARAM = {"name": "message_id", "in": "path", "required": True, "schema": {"type": "integer"}}
956982
TAG_SLUG_PARAM = {"name": "tag_slug", "in": "path", "required": True, "schema": {"type": "string"}}
957983
LIST_KEY_PARAM = {"name": "list_key", "in": "path", "required": True, "schema": {"type": "string"}}
984+
JOB_ID_PARAM = {"name": "job_id", "in": "path", "required": True, "schema": {"type": "integer"}}
958985
CAMPAIGN_EXTERNAL_KEY_PARAM = {
959986
"name": "external_key",
960987
"in": "path",
@@ -1366,6 +1393,36 @@ def bearer_responses(success, *, accepted=False):
13661393
),
13671394
}
13681395
},
1396+
"/api/recipient-lists/{list_key}/imports": {
1397+
"post": {
1398+
"tags": ["Recipient Lists"],
1399+
"summary": "Create recipient list import job",
1400+
"description": "Creates a pending job that fetches JSONL member rows from a client-owned URL. Process the job asynchronously, then poll the job status before dependent sends.",
1401+
"security": [{"BearerAuth": []}],
1402+
"parameters": [LIST_KEY_PARAM],
1403+
"requestBody": json_body("#/components/schemas/RecipientListImportRequest"),
1404+
"responses": bearer_responses(
1405+
json_response("Recipient list import job", "#/components/schemas/RecipientListImportJobResponse"),
1406+
accepted=True,
1407+
),
1408+
}
1409+
},
1410+
"/api/recipient-lists/{list_key}/imports/{job_id}": {
1411+
"get": {
1412+
"tags": ["Recipient Lists"],
1413+
"summary": "Get recipient list import job",
1414+
"security": [{"BearerAuth": []}],
1415+
"parameters": [
1416+
LIST_KEY_PARAM,
1417+
JOB_ID_PARAM,
1418+
{"name": "audience", "in": "query", "required": True, "schema": {"type": "string"}},
1419+
{"name": "client", "in": "query", "required": True, "schema": {"type": "string"}},
1420+
],
1421+
"responses": bearer_responses(
1422+
json_response("Recipient list import job", "#/components/schemas/RecipientListImportJobResponse")
1423+
),
1424+
}
1425+
},
13691426
"/api/recipient-lists/{list_key}/transactional-send": {
13701427
"post": {
13711428
"tags": ["Recipient Lists"],
@@ -1823,6 +1880,10 @@ def bearer_responses(success, *, accepted=False):
18231880
"properties": {"deleted_count": {"type": "integer"}},
18241881
},
18251882
"RecipientListType": {"type": "string", "enum": [choice.value for choice in RecipientListType]},
1883+
"RecipientListImportJobStatus": {
1884+
"type": "string",
1885+
"enum": [choice.value for choice in RecipientListImportJobStatus],
1886+
},
18261887
"RecipientListInput": {
18271888
"type": "object",
18281889
"properties": {
@@ -1887,6 +1948,21 @@ def bearer_responses(success, *, accepted=False):
18871948
},
18881949
]
18891950
},
1951+
"RecipientListImportRequest": {
1952+
"allOf": [
1953+
{"$ref": "#/components/schemas/ScopedMutationRequest"},
1954+
{
1955+
"type": "object",
1956+
"required": ["source_url"],
1957+
"properties": {
1958+
"source_url": {"type": "string", "format": "uri"},
1959+
"idempotency_key": {"type": "string", "maxLength": 255},
1960+
"list": {"$ref": "#/components/schemas/RecipientListInput"},
1961+
"remove_absent": {"type": "boolean"},
1962+
},
1963+
},
1964+
]
1965+
},
18901966
"RecipientListTransactionalSendRequest": {
18911967
"allOf": [
18921968
{"$ref": "#/components/schemas/ScopedMutationRequest"},
@@ -2268,6 +2344,45 @@ def bearer_responses(success, *, accepted=False):
22682344
"removed_count": {"type": "integer"},
22692345
},
22702346
},
2347+
"RecipientListImportJob": {
2348+
"type": "object",
2349+
"properties": {
2350+
"id": {"type": "integer"},
2351+
"list_key": {"type": "string"},
2352+
"audience": {"type": "string"},
2353+
"client": {"type": "string"},
2354+
"source_url": {"type": "string", "format": "uri"},
2355+
"idempotency_key": {"type": "string"},
2356+
"status": {"$ref": "#/components/schemas/RecipientListImportJobStatus"},
2357+
"list": {"type": "object"},
2358+
"remove_absent": {"type": "boolean"},
2359+
"row_count": {"type": "integer"},
2360+
"created_count": {"type": "integer"},
2361+
"updated_count": {"type": "integer"},
2362+
"removed_count": {"type": "integer"},
2363+
"failed_count": {"type": "integer"},
2364+
"failed_rows": {"type": "array", "items": {"type": "object"}},
2365+
"content_sha256": {"type": "string"},
2366+
"error": {"type": "string"},
2367+
"started_at": {"type": ["string", "null"], "format": "date-time"},
2368+
"completed_at": {"type": ["string", "null"], "format": "date-time"},
2369+
"created_at": {"type": "string", "format": "date-time"},
2370+
"updated_at": {"type": "string", "format": "date-time"},
2371+
"recipient_list": {
2372+
"anyOf": [
2373+
{"$ref": "#/components/schemas/RecipientList"},
2374+
{"type": "null"},
2375+
]
2376+
},
2377+
},
2378+
},
2379+
"RecipientListImportJobResponse": {
2380+
"type": "object",
2381+
"properties": {
2382+
"import_job": {"$ref": "#/components/schemas/RecipientListImportJob"},
2383+
"created": {"type": "boolean"},
2384+
},
2385+
},
22712386
"RecipientListTransactionalSendResponse": {
22722387
"type": "object",
22732388
"properties": {

0 commit comments

Comments
 (0)