feat: added advertisers endpoints

2025-02-16 12:32:11 +03:00
parent 7013635988
commit bdafd96d94
4 changed files with 302 additions and 0 deletions
@@ -0,0 +1,21 @@
+import typing
+from uuid import UUID
+
+from ninja import ModelSchema
+
+from apps.advertiser.models import Advertiser as AdvertiserModel
+from apps.mlscore.models import Mlscore as MlscoreModel
+
+
+class Advertiser(ModelSchema):
+    advertiser_id: UUID
+
+    class Meta:
+        model = AdvertiserModel
+        exclude: typing.ClassVar[tuple[str]] = (AdvertiserModel.id.field.name,)
+
+
+class Mlscore(ModelSchema):
+    class Meta:
+        model = MlscoreModel
+        exclude: typing.ClassVar[tuple[str]] = (MlscoreModel.id.field.name,)
@@ -0,0 +1,194 @@
+import json
+import uuid
+from http import HTTPStatus as status
+
+from django.test import TestCase, Client
+from apps.advertiser.models import Advertiser
+from apps.client.models import Client as ClientModel
+from apps.mlscore.models import Mlscore
+
+
+class TestMlscoreEndpoint(TestCase):
+    def setUp(self):
+        self.client = Client()
+        self.advertiser = Advertiser.objects.create(name="Test Advertiser")
+        self.client_obj = ClientModel.objects.create(
+            login="test_client",
+            age=14,
+            location="test_location",
+            gender=ClientModel.GenderChoices.FEMALE,
+        )
+
+        self.url = "/advertisers/ml-scores"
+
+    def test_create_mlscore_success(self):
+        data = {
+            "advertiser_id": str(self.advertiser.id),
+            "client_id": str(self.client_obj.id),
+            "score": 90,
+        }
+        response = self.client.post(
+            self.url, json.dumps(data), content_type="application/json"
+        )
+        mlscore = Mlscore.objects.first()
+
+        self.assertEqual(response.status_code, status.OK)
+        self.assertEqual(mlscore.score, 90)
+
+    def test_update_mlscore_success(self):
+        mlscore = Mlscore.objects.create(
+            advertiser=self.advertiser,
+            client=self.client_obj,
+            score=80,
+        )
+        data = {
+            "advertiser_id": str(self.advertiser.id),
+            "client_id": str(self.client_obj.id),
+            "score": 85,
+        }
+        response = self.client.post(
+            self.url, json.dumps(data), content_type="application/json"
+        )
+        mlscore.refresh_from_db()
+
+        self.assertEqual(response.status_code, status.OK)
+        self.assertEqual(mlscore.score, 85)
+
+    def test_missing_required_field(self):
+        data = {"score": 90}
+        response = self.client.post(
+            self.url, json.dumps(data), content_type="application/json"
+        )
+
+        self.assertEqual(response.status_code, status.BAD_REQUEST)
+
+    def test_invalid_uuid_format(self):
+        data = {
+            "advertiser_id": "invalid-uuid",
+            "client_id": str(self.client_obj.id),
+            "score": 90,
+        }
+        response = self.client.post(
+            self.url, json.dumps(data), content_type="application/json"
+        )
+
+        self.assertEqual(response.status_code, status.BAD_REQUEST)
+
+    def test_non_existing_client(self):
+        data = {
+            "advertiser_id": str(self.advertiser.id),
+            "client_id": str(uuid.uuid4()),
+            "score": 90,
+        }
+        response = self.client.post(
+            self.url, json.dumps(data), content_type="application/json"
+        )
+
+        self.assertEqual(response.status_code, status.BAD_REQUEST)
+
+    def test_non_existing_advertiser(self):
+        data = {
+            "advertiser_id": str(uuid.uuid4()),
+            "client_id": str(self.client_obj.id),
+            "score": 90,
+        }
+        response = self.client.post(
+            self.url, json.dumps(data), content_type="application/json"
+        )
+
+        self.assertEqual(response.status_code, status.BAD_REQUEST)
+
+
+class TestBulkAdvertisersEndpoint(TestCase):
+    def setUp(self):
+        self.client = Client()
+        self.url = "/advertisers/bulk"
+        self.advertiser = Advertiser.objects.create(name="Advertiser 1")
+
+    def test_bulk_create_success(self):
+        uuid1 = self.advertiser.id
+        uuid2 = uuid.uuid4()
+        data = [
+            {"advertiser_id": str(uuid1), "name": "Advertiser 4"},
+            {"advertiser_id": str(uuid2), "name": "Advertiser 1"},
+            {"advertiser_id": str(uuid2), "name": "Advertiser 5"},
+            {"advertiser_id": str(uuid2), "name": "Advertiser 2"},
+            {"advertiser_id": str(uuid1), "name": "Advertiser 2"},
+        ]
+        response = self.client.post(
+            self.url, json.dumps(data), content_type="application/json"
+        )
+        self.advertiser.refresh_from_db()
+
+        self.assertEqual(response.status_code, status.CREATED)
+        self.assertEqual(self.advertiser.name, "Advertiser 2")
+        self.assertEqual(Advertiser.objects.count(), 2)
+
+    def test_bulk_update_success(self):
+        advertiser = Advertiser.objects.create(name="Old Name")
+        data = [{"advertiser_id": str(advertiser.id), "name": "New Name"}]
+        response = self.client.post(
+            self.url, json.dumps(data), content_type="application/json"
+        )
+        advertiser.refresh_from_db()
+
+        self.assertEqual(response.status_code, status.CREATED)
+        self.assertEqual(advertiser.name, "New Name")
+
+    def test_duplicate_advertiser_ids(self):
+        adv_id = uuid.uuid4()
+        data = [
+            {"advertiser_id": str(adv_id), "name": "First"},
+            {"advertiser_id": str(adv_id), "name": "Last"},
+        ]
+        response = self.client.post(
+            self.url, json.dumps(data), content_type="application/json"
+        )
+        advertiser = Advertiser.objects.get(id=adv_id)
+
+        self.assertEqual(response.status_code, status.CREATED)
+        self.assertEqual(advertiser.name, "Last")
+
+    def test_invalid_advertiser_id_format(self):
+        data = [{"advertiser_id": "invalid", "name": "Invalid"}]
+        response = self.client.post(
+            self.url, json.dumps(data), content_type="application/json"
+        )
+
+        self.assertEqual(response.status_code, status.BAD_REQUEST)
+
+    def test_empty_bulk_request(self):
+        response = self.client.post(
+            self.url, json.dumps([]), content_type="application/json"
+        )
+
+        self.assertEqual(response.status_code, status.CREATED)
+        self.assertEqual(len(response.json()), 0)
+
+
+class TestGetAdvertiserEndpoint(TestCase):
+    def setUp(self):
+        self.client = Client()
+        self.advertiser = Advertiser.objects.create(name="Test Advertiser")
+        self.url = "/advertisers"
+        self.valid_url = f"{self.url}/{self.advertiser.id}"
+
+    def test_get_advertiser_success(self):
+        response = self.client.get(self.valid_url)
+
+        self.assertEqual(response.status_code, status.OK)
+        self.assertEqual(
+            response.json()["advertiser_id"], str(self.advertiser.id)
+        )
+        self.assertEqual(response.json()["name"], self.advertiser.name)
+
+    def test_non_existent_advertiser(self):
+        non_existent_url = f"{self.url}/{uuid.uuid4()}"
+        response = self.client.get(non_existent_url)
+
+        self.assertEqual(response.status_code, status.NOT_FOUND)
+
+    def test_invalid_uuid_format(self):
+        response = self.client.get(f"{self.url}/invalid-uuid")
+
+        self.assertEqual(response.status_code, status.BAD_REQUEST)
@@ -0,0 +1,87 @@
+from collections import defaultdict
+from http import HTTPStatus as status
+from uuid import UUID
+
+from django.db import transaction
+from django.http import HttpRequest
+from django.shortcuts import get_object_or_404
+from ninja import Router
+
+from api.v1 import schemas as global_schemas
+from api.v1.advertisers import schemas
+from apps.advertiser.models import Advertiser
+from apps.mlscore.models import Mlscore
+
+router = Router(tags=["advertisers"])
+
+
+@router.post(
+    "/ml-scores",
+    response={
+        status.OK: schemas.Mlscore,
+        status.BAD_REQUEST: global_schemas.BadRequestError,
+    },
+)
+def create_or_update_mlscore(
+    request: HttpRequest, mlscore: schemas.Mlscore
+) -> tuple[status, schemas.Mlscore]:
+    mlscore_instance, _ = Mlscore.objects.update_or_create(
+        advertiser_id=mlscore.advertiser,
+        client_id=mlscore.client,
+        defaults={"score": mlscore.score},
+    )
+
+    return status.OK, mlscore_instance
+
+
+@router.post(
+    "/bulk",
+    response={
+        status.CREATED: list[schemas.Advertiser],
+        status.BAD_REQUEST: global_schemas.BadRequestError,
+    },
+)
+def bulk_create_or_update(
+    request: HttpRequest, data: list[schemas.Advertiser]
+) -> tuple[status, list[Advertiser]]:
+    latest_advertisers: dict[UUID, schemas.Advertiser] = defaultdict(
+        lambda: None
+    )
+
+    for item in reversed(data):
+        if latest_advertisers[item.advertiser_id] is None:
+            Advertiser(
+                id=item.advertiser_id, **item.dict(exclude={"client_id"})
+            ).validate(
+                validate_unique=False,
+                validate_constraints=False,
+            )
+            latest_advertisers[item.advertiser_id] = item
+
+    unique_advertisers = reversed(list(latest_advertisers.values()))
+
+    result = []
+
+    with transaction.atomic():
+        for advertiser in unique_advertisers:
+            advertiser_instance, _ = Advertiser.objects.update_or_create(
+                id=advertiser.advertiser_id,
+                defaults={**dict(advertiser)},
+            )
+            result.append(advertiser_instance)
+
+    return status.CREATED, result
+
+
+@router.get(
+    "/{advertiser_id}",
+    response={
+        status.OK: schemas.Advertiser,
+        status.BAD_REQUEST: global_schemas.BadRequestError,
+        status.NOT_FOUND: global_schemas.NotFoundError,
+    },
+)
+def get_advertiser(
+    request: HttpRequest, advertiser_id: UUID
+) -> tuple[status, Advertiser]:
+    return status.OK, get_object_or_404(Advertiser, id=advertiser_id)