feat: added promocode creation and view

solution/api/v1/auth.py

@@ -7,6 +7,7 @@ from ninja.security import HttpBearer
 from pydantic import BaseModel, ValidationError
 
 import apps.business.models
+import apps.user.models
 
 
 class BusinessToken(BaseModel):
@@ -28,7 +29,7 @@ class BusinessAuth(HttpBearer):
 
         try:
             business = apps.business.models.Business.objects.get(
-                id=token_payload.business_id
+                id=token_payload.business_id,
             )
         except apps.business.models.Business.DoesNotExist:
             return None
@@ -37,3 +38,31 @@ class BusinessAuth(HttpBearer):
             return None
 
         return business
+
+
+class UserToken(BaseModel):
+    user_id: uuid.UUID
+    token_version: int
+
+
+class UserAuth(HttpBearer):
+    def authenticate(self, request: HttpRequest, token: str) -> str | None:
+        try:
+            decoded_payload = jwt.decode(
+                token,
+                settings.SECRET_KEY,
+                algorithms=["HS256"],
+            )
+            token_payload = UserToken(**decoded_payload)
+        except (jwt.PyJWTError, ValidationError):
+            return None
+
+        try:
+            user = apps.user.models.User.objects.get(id=token_payload.user_id)
+        except apps.user.models.User.DoesNotExist:
+            return None
+
+        if user.token_version != token_payload.token_version:
+            return None
+
+        return user

solution/api/v1/business/schemas.py

@@ -1,20 +1,20 @@
-import re
+import datetime
 import uuid
-from typing import ClassVar
+from typing import ClassVar, Literal
 
 from ninja import ModelSchema, Schema
-from pydantic import EmailStr, field_validator
+from pydantic import Field
 
 from apps.business.models import Business
+from apps.promo.models import Promocode, PromocodeTarget
 
 
 class BusinessSignUpIn(ModelSchema):
-    email: EmailStr
-
     class Meta:
         model = Business
         fields: ClassVar[list[str]] = [
             Business.name.field.name,
+            Business.email.field.name,
             Business.password.field.name,
         ]
 
@@ -24,27 +24,85 @@ class BusinessSignUpOut(Schema):
     company_id: uuid.UUID
 
 
-class BusinessSignInIn(Schema):
-    email: EmailStr
-    password: str
-
-    @field_validator("password")
-    def validate_password(cls, value: str) -> str:  # noqa: N805
-        pattern = r"^(?=.*[a-z])(?=.*[A-Z])(?=.*\d)(?=.*[@$!%*?&])[A-Za-z\d@$!%*?&]{8,60}$"  # noqa: E501
-        if not re.match(pattern, value):
-            e = (
-                "Password must contain at least 8 characters, one uppercase "
-                "letter, one lowercase letter, one number, and one special "
-                "character (@$!%*?&)."
-            )
-            raise ValueError(e)
-        return value
+class BusinessSignInIn(ModelSchema):
+    class Meta:
+        model = Business
+        fields: ClassVar[list[str]] = [
+            Business.email.field.name,
+            Business.password.field.name,
+        ]
 
 
 class BusinessSignInOut(Schema):
     token: str
 
 
+class PromocodeTarget(ModelSchema):
+    categories: list[str] | None = None
+
+    class Meta:
+        model = PromocodeTarget
+        fields: ClassVar[list[str]] = [
+            PromocodeTarget.age_from.field.name,
+            PromocodeTarget.age_until.field.name,
+            PromocodeTarget.country.field.name,
+            PromocodeTarget.categories.field.name,
+        ]
+
+
+class CreatePromocodeIn(ModelSchema):
+    target: PromocodeTarget
+    promo_unique: list[str] | None = None
+
+    class Meta:
+        model = Promocode
+        fields: ClassVar[list[str]] = [
+            Promocode.description.field.name,
+            Promocode.image_url.field.name,
+            Promocode.max_count.field.name,
+            Promocode.active_from.field.name,
+            Promocode.active_until.field.name,
+            Promocode.mode.field.name,
+            Promocode.promo_common.field.name,
+        ]
+
+
+class CreatePromocodeOut(Schema):
+    id: uuid.UUID
+
+
+class PromocodeListFilters(Schema):
+    limit: int = 10
+    offset: int = 0
+    query: Literal["active_from", "active_until", None] = None
+    country__in: list[str] = Field(None, alias="country")
+
+
+class PromocodeTargetViewOut(Schema):
+    age_from: int | None
+    age_until: int | None
+    country: str | None
+    categories: list[str] | None
+
+
+class PromocodeViewOut(Schema):
+    promo_id: uuid.UUID
+    company_id: uuid.UUID
+    description: str
+    image_url: str | None
+    target: PromocodeTargetViewOut
+    max_count: int
+    active_from: datetime.date | None
+    active_until: datetime.date | None
+    mode: str
+    promo_common: str | None
+    promo_unique: list[str] | None
+    company_name: str
+    like_count: int
+    used_count: int
+    active: bool
+
+
 __all__ = [
     "BusinessSignInIn",
     "BusinessSignInOut",

solution/api/v1/business/views.py

@@ -1,12 +1,18 @@
+import datetime
 from http import HTTPStatus as status
 
-from django.http import HttpRequest
-from ninja import Router
-from ninja.errors import AuthenticationError
+from django.core.exceptions import ValidationError
+from django.db.models import Count, Q, Value
+from django.db.models.functions import Coalesce
+from django.http import HttpRequest, HttpResponse
+from ninja import Query, Router
+from ninja.errors import AuthenticationError, HttpError
 
 from api.v1 import schemas as global_schemas
+from api.v1.auth import BusinessAuth
 from api.v1.business import schemas
 from apps.business.models import Business
+from apps.promo.models import Promocode, PromocodeTarget
 
 router = Router(tags=["business"])
 
@@ -20,13 +26,15 @@ router = Router(tags=["business"])
     },
 )
 def signup(
-    request: HttpRequest, business: schemas.BusinessSignUpIn
+    request: HttpRequest,
+    business: schemas.BusinessSignUpIn,
 ) -> tuple[int, schemas.BusinessSignUpOut]:
     business_obj = Business(**business.dict())
     business_obj.save()
 
     return status.OK, schemas.BusinessSignUpOut(
-        token=business_obj.generate_token(), company_id=business_obj.id
+        token=business_obj.generate_token(),
+        company_id=business_obj.id,
     )
 
 
@@ -39,8 +47,16 @@ def signup(
     },
 )
 def signin(
-    request: HttpRequest, login_data: schemas.BusinessSignInIn
+    request: HttpRequest,
+    login_data: schemas.BusinessSignInIn,
 ) -> tuple[int, schemas.BusinessSignInOut]:
+    business_obj = Business(**dict(login_data))
+    business_obj.validate(
+        include=[Business.email.field, Business.password.field],
+        validate_unique=False,
+        validate_constraints=False,
+    )
+
     try:
         business_obj = Business.objects.get(email=login_data.email)
     except Business.DoesNotExist:
@@ -53,5 +69,167 @@ def signin(
     business_obj.save()
 
     return status.OK, schemas.BusinessSignInOut(
-        token=business_obj.generate_token()
+        token=business_obj.generate_token(),
+    )
+
+
+@router.post(
+    "/promo",
+    auth=BusinessAuth(),
+    response={
+        status.CREATED: schemas.CreatePromocodeOut,
+        status.BAD_REQUEST: global_schemas.ValidationError,
+        status.UNAUTHORIZED: global_schemas.UnauthorizedError,
+    },
+)
+def create_promocode(
+    request: HttpRequest,
+    promocode: schemas.CreatePromocodeIn,
+) -> schemas.CreatePromocodeOut:
+    business = request.auth
+    promocode = dict(promocode)
+    target = dict(promocode.pop("target"))
+
+    target_obj = PromocodeTarget(**target)
+    target_obj.save()
+
+    promocode_obj = Promocode(
+        business=business,
+        target=target_obj,
+        **promocode,
+    )
+    try:
+        promocode_obj.save()
+    except ValidationError as e:
+        target_obj.delete()
+        raise e  # noqa: TRY201
+
+    return status.CREATED, schemas.CreatePromocodeOut(id=promocode_obj.id)
+
+
+@router.get(
+    "/promo",
+    auth=BusinessAuth(),
+    response={
+        status.OK: list[schemas.PromocodeViewOut],
+        status.BAD_REQUEST: global_schemas.ValidationError,
+    },
+    exclude_none=True,
+)
+def list_promocode(
+    request: HttpRequest,
+    filters: Query[schemas.PromocodeListFilters],
+    response: HttpResponse,
+) -> list[schemas.PromocodeViewOut]:
+    business = request.auth
+
+    promocodes = Promocode.objects.filter(business=business)
+
+    if filters.country__in:
+        promocodes = promocodes.filter(
+            Q(target__country__in=filters.country__in)
+            | Q(target__country__isnull=True)
+        )
+
+    if filters.query == "active_from":
+        promocodes = promocodes.annotate(
+            active_from_sort=Coalesce("active_from", Value(datetime.min))
+        ).order_by("-active_from_sort")
+    elif filters.query == "active_until":
+        promocodes = promocodes.annotate(
+            active_until_sort=Coalesce("active_until", Value(datetime.max))
+        ).order_by("-active_until_sort")
+    else:
+        promocodes = promocodes.order_by("-created_at")
+
+    promocodes = promocodes.annotate(
+        used_count=Count("activations"),
+        like_count=Count("comments"),
+    )
+
+    response["X-Total-Count"] = promocodes.count()
+
+    promocodes = promocodes[filters.offset : filters.offset + filters.limit]
+
+    return [
+        schemas.PromocodeViewOut(
+            promo_id=promocode.id,
+            company_id=promocode.business.id,
+            company_name=promocode.business.name,
+            description=promocode.description,
+            image_url=promocode.image_url,
+            target=schemas.PromocodeTargetViewOut(
+                age_from=promocode.target.age_from,
+                age_until=promocode.target.age_until,
+                country=promocode.target.country.code
+                if promocode.target.country
+                else None,
+                categories=promocode.target.categories,
+            ),
+            max_count=promocode.max_count,
+            active_from=promocode.active_from,
+            active_until=promocode.active_until,
+            mode=promocode.mode,
+            promo_common=promocode.promo_common,
+            promo_unique=promocode.promo_unique,
+            like_count=promocode.like_count,
+            used_count=promocode.used_count,
+            active=promocode.active,
+        )
+        for promocode in promocodes
+    ]
+
+
+@router.get(
+    "/promo/{promocode_id}",
+    auth=BusinessAuth(),
+    response={
+        status.OK: schemas.PromocodeViewOut,
+        status.NOT_FOUND: global_schemas.NotFoundError,
+    },
+    exclude_none=True,
+)
+def product_get(
+    request: HttpRequest, promocode_id: str
+) -> schemas.PromocodeViewOut:
+    business = request.auth
+
+    promocodes = Promocode.objects.filter(id=promocode_id)
+
+    if len(promocodes) == 0:
+        raise HttpError(status.NOT_FOUND, status.NOT_FOUND.phrase)
+
+    promocodes = promocodes.annotate(
+        used_count=Count("activations"),
+        like_count=Count("comments"),
+    )
+
+    promocode = promocodes[0]
+
+    if promocode.business != business:
+        raise HttpError(status.FORBIDDEN, status.FORBIDDEN.phrase)
+
+    return schemas.PromocodeViewOut(
+        promo_id=promocode.id,
+        company_id=promocode.business.id,
+        company_name=promocode.business.name,
+        description=promocode.description,
+        image_url=promocode.image_url,
+        target=schemas.PromocodeTargetViewOut(
+            age_from=promocode.target.age_from,
+            age_until=promocode.target.age_until,
+            country=promocode.target.country.code
+            if promocode.target.country
+            else None,
+            categories=promocode.target.categories,
+        ),
+        max_count=promocode.max_count,
+        active_from=promocode.active_from,
+        active_until=promocode.active_until,
+        mode=promocode.mode,
+        promo_common=promocode.promo_common,
+        promo_unique=promocode.promo_unique,
+        like_count=promocode.like_count,
+        used_count=promocode.used_count,
+        active=promocode.active,
     )

solution/api/v1/handlers.py

@@ -46,7 +46,9 @@ def handle_django_validation_error(
 
 
 def handle_authentication_error(
-    request: HttpRequest, exc: errors.AuthenticationError, router: NinjaAPI
+    request: HttpRequest,
+    exc: errors.AuthenticationError,
+    router: NinjaAPI,
 ) -> HttpResponse:
     return router.create_response(
         request,
@@ -56,7 +58,9 @@ def handle_authentication_error(
 
 
 def handle_validation_error(
-    request: HttpRequest, exc: errors.ValidationError, router: NinjaAPI
+    request: HttpRequest,
+    exc: errors.ValidationError,
+    router: NinjaAPI,
 ) -> HttpResponse:
     return router.create_response(
         request,
@@ -66,7 +70,9 @@ def handle_validation_error(
 
 
 def handle_not_found_error(
-    request: HttpRequest, exc: Exception, router: NinjaAPI
+    request: HttpRequest,
+    exc: Exception,
+    router: NinjaAPI,
 ) -> HttpResponse:
     return router.create_response(
         request,
@@ -76,7 +82,9 @@ def handle_not_found_error(
 
 
 def handle_unknown_exception(
-    request: HttpRequest, exc: Exception, router: NinjaAPI
+    request: HttpRequest,
+    exc: Exception,
+    router: NinjaAPI,
 ) -> HttpResponse:
     logger.exception(exc)