schon/engine/payments/managers.py
Egor fureunoir Gorbunov ad320235d6 feat(payments, vibes_auth, core): introduce decimal fields, 2FA, and admin OTP
- Refactored monetary fields across models to use `DecimalField` for improved precision.
- Implemented two-factor authentication (2FA) for admin logins with OTP codes.
- Added ability to generate admin OTP via management commands.
- Updated Docker Compose override for dev-specific port bindings.
- Included template for 2FA OTP verification to enhance security.

Additional changes:
- Upgraded and downgraded various dependencies (e.g., django-celery-beat and yarl).
- Replaced float-based calculations with decimal for consistent rounding behavior.
- Improved admin user management commands for activation and OTP generation.
2026-03-03 00:42:21 +03:00

66 lines
2.1 KiB
Python

from django.db.models import (
BooleanField,
Case,
DecimalField,
F,
Manager,
Q,
QuerySet,
Sum,
Value,
When,
)
from django.db.models.functions import Coalesce
from django.utils.timezone import now
class GatewayQuerySet(QuerySet):
def with_usage_sums(self) -> QuerySet:
today = now().date()
current_month_start = today.replace(day=1)
return self.annotate(
daily_sum=Coalesce(
Sum(
"transactions__amount", filter=Q(transactions__created__date=today)
),
Value(0),
output_field=DecimalField(max_digits=12, decimal_places=2),
),
monthly_sum=Coalesce(
Sum(
"transactions__amount",
filter=Q(transactions__created__date__gte=current_month_start),
),
Value(0),
output_field=DecimalField(max_digits=12, decimal_places=2),
),
)
def can_be_used(self) -> QuerySet:
qs = self.with_usage_sums()
qs = qs.annotate(
daily_ok=Case(
When(daily_limit=0, then=Value(True)),
When(daily_sum__lt=F("daily_limit"), then=Value(True)),
default=Value(False),
output_field=BooleanField(),
),
monthly_ok=Case(
When(monthly_limit=0, then=Value(True)),
When(monthly_sum__lt=F("monthly_limit"), then=Value(True)),
default=Value(False),
output_field=BooleanField(),
),
)
return qs.annotate(
can_be_used=Case(
When(daily_ok=True, monthly_ok=True, is_active=True, then=Value(True)),
default=Value(False),
output_field=BooleanField(),
)
).order_by("-priority")
class GatewayManager(Manager.from_queryset(GatewayQuerySet)): # ty:ignore[unsupported-base]
def get_queryset(self) -> QuerySet:
return super().get_queryset().can_be_used()