diff --git a/backend/api/admin/api_keys.py b/backend/api/admin/api_keys.py index 19e501fc..8b5f8b3b 100644 --- a/backend/api/admin/api_keys.py +++ b/backend/api/admin/api_keys.py @@ -1,14 +1,15 @@ from django.contrib import messages -from django.http import HttpRequest, HttpResponseBadRequest, HttpResponse +from django.http import HttpResponseBadRequest, HttpResponse from django.shortcuts import render, redirect from backend.models import APIKey +from backend.types.htmx import HtmxHttpRequest # Still working on -def generate_api_key(request: HttpRequest) -> HttpResponse(): +def generate_api_key(request: HtmxHttpRequest) -> HttpResponse: if not request.htmx: return redirect("user settings") if not request.user.is_staff or not request.user.is_superuser: diff --git a/backend/api/base/modal.py b/backend/api/base/modal.py index ac16de97..c63f12a4 100644 --- a/backend/api/base/modal.py +++ b/backend/api/base/modal.py @@ -1,7 +1,6 @@ from __future__ import annotations from django.contrib import messages -from django.http import HttpRequest from django.http import HttpResponseBadRequest from django.shortcuts import render @@ -10,6 +9,7 @@ from backend.models import QuotaLimit from backend.models import Team from backend.models import UserSettings +from backend.types.htmx import HtmxHttpRequest from backend.utils.feature_flags import get_feature_status from backend.utils.quota_limit_ops import quota_usage_check_under @@ -17,7 +17,7 @@ # Still working on -def open_modal(request: HttpRequest, modal_name, context_type=None, context_value=None): +def open_modal(request: HtmxHttpRequest, modal_name, context_type=None, context_value=None): try: context = {} template_name = f"modals/{modal_name}.html" diff --git a/backend/api/base/notifications.py b/backend/api/base/notifications.py index 7f027082..5fcb1f2a 100644 --- a/backend/api/base/notifications.py +++ b/backend/api/base/notifications.py @@ -1,10 +1,11 @@ -from django.http import HttpRequest, HttpResponse +from django.http import HttpResponse from django.shortcuts import render from backend.models import Notification +from backend.types.htmx import HtmxHttpRequest -def get_notification_html(request: HttpRequest): +def get_notification_html(request: HtmxHttpRequest): user_notifications = Notification.objects.filter(user=request.user).order_by("-date") above_5 = False @@ -19,7 +20,7 @@ def get_notification_html(request: HttpRequest): ) -def delete_notification(request: HttpRequest, id: int): +def delete_notification(request: HtmxHttpRequest, id: int): notif = Notification.objects.filter(id=id, user=request.user).first() if notif is None or notif.user != request.user: diff --git a/backend/api/clients/fetch.py b/backend/api/clients/fetch.py index 6f58e4ce..40966ac4 100644 --- a/backend/api/clients/fetch.py +++ b/backend/api/clients/fetch.py @@ -1,13 +1,13 @@ from django.db.models import Q -from django.http import HttpRequest from django.shortcuts import render, redirect from django.views.decorators.http import require_http_methods from backend.models import Client +from backend.types.htmx import HtmxHttpRequest @require_http_methods(["GET"]) -def fetch_all_clients(request: HttpRequest): +def fetch_all_clients(request: HtmxHttpRequest): if not request.htmx: return redirect("clients dashboard") @@ -25,7 +25,7 @@ def fetch_all_clients(request: HttpRequest): @require_http_methods(["GET"]) -def fetch_clients_dropdown(request: HttpRequest): +def fetch_clients_dropdown(request: HtmxHttpRequest): if not request.htmx: return redirect("clients dashboard") diff --git a/backend/api/currency_converter/convert.py b/backend/api/currency_converter/convert.py index a17f9e06..bfa3e0fe 100644 --- a/backend/api/currency_converter/convert.py +++ b/backend/api/currency_converter/convert.py @@ -1,11 +1,12 @@ import datetime from django.contrib import messages -from django.http import HttpResponse, HttpRequest +from django.http import HttpResponse from django.shortcuts import render, redirect from forex_python.converter import CurrencyRates -from backend.models import * +from backend.models import UserSettings +from backend.types.htmx import HtmxHttpRequest def convert_currency(init_currency, target_currency, amount, date=None): @@ -64,7 +65,7 @@ def convert_currency(init_currency, target_currency, amount, date=None): raise ValueError(f"Error in currency conversion: {e}") -def currency_conversion(request: HttpRequest): +def currency_conversion(request: HtmxHttpRequest): context = {} if not request.htmx: @@ -73,14 +74,14 @@ def currency_conversion(request: HttpRequest): if request.method != "POST": return HttpResponse("Method not allowed", status=405) - amount = request.POST.get("currency_amount") + amount_str: str = request.POST.get("currency_amount", "") - if not amount or len(amount) > 10: + if not amount_str or len(amount_str) > 10: messages.error(request, "Please enter a valid currency amount") return render(request, "partials/messages_list.html") try: - amount = float(amount) + amount: float = float(amount_str) converted_amt = convert_currency( request.POST["from_currency"], request.POST["to_currency"], @@ -100,6 +101,6 @@ def currency_conversion(request: HttpRequest): } ) return render(request, "pages/currency_converter/result.html", context) - except Exception as e: + except (KeyError, ValueError, TypeError, AttributeError): messages.error(request, f"Failed to convert currency. Make sure the amount is valid.") return render(request, "partials/messages_list.html") diff --git a/backend/api/emails/fetch.py b/backend/api/emails/fetch.py index 0f1ce715..7c6c9bde 100644 --- a/backend/api/emails/fetch.py +++ b/backend/api/emails/fetch.py @@ -1,15 +1,16 @@ from django.contrib import messages -from django.core.paginator import Paginator -from django.db.models import Q -from django.http import HttpRequest, HttpResponse +from django.core.paginator import Paginator, Page +from django.db.models import Q, QuerySet +from django.http import HttpResponse from django.shortcuts import render, redirect from django_ratelimit.core import is_ratelimited from django_ratelimit.decorators import ratelimit from backend.models import EmailSendStatus +from backend.types.htmx import HtmxHttpRequest -def fetch_all_emails(request: HttpRequest): +def fetch_all_emails(request: HtmxHttpRequest): if is_ratelimited(request, group="fetch_all_emails", key="user", rate="2/4s", increment=True) or is_ratelimited( request, group="fetch_all_emails", @@ -26,17 +27,17 @@ def fetch_all_emails(request: HttpRequest): page_num = request.GET.get("page") if request.user.logged_in_as_team: - results = EmailSendStatus.objects.filter(organization=request.user.logged_in_as_team) + results: QuerySet[EmailSendStatus] = EmailSendStatus.objects.filter(organization=request.user.logged_in_as_team) else: results = EmailSendStatus.objects.filter(user=request.user) if search_text: results = results.filter(Q(recipient__icontains=search_text)) - results = results.order_by("-id") + result: Page | QuerySet = results.order_by("-id") - paginator = Paginator(results, 8) - results = paginator.get_page(page_num) + paginator = Paginator(result, 8) + result = paginator.get_page(page_num) - context.update({"emails": results}) + context.update({"emails": result}) return render(request, "pages/emails/_fetch_body.html", context) diff --git a/backend/api/emails/send.py b/backend/api/emails/send.py index e8236c68..1e14b8e9 100644 --- a/backend/api/emails/send.py +++ b/backend/api/emails/send.py @@ -30,6 +30,7 @@ ) from backend.utils.quota_limit_ops import quota_usage_check_under from settings.helpers import send_email, send_templated_bulk_email +from backend.types.htmx import HtmxHttpRequest @dataclass @@ -44,7 +45,7 @@ class Invalid: @require_POST @htmx_only("emails:dashboard") @feature_flag_check("areUserEmailsAllowed", status=True, api=True, htmx=True) -def send_single_email_view(request: WSGIRequest) -> HttpResponse: +def send_single_email_view(request: HtmxHttpRequest) -> HttpResponse: check_usage = quota_usage_check_under(request, "emails-single-count", api=True, htmx=True) if not isinstance(check_usage, bool): return check_usage @@ -55,7 +56,7 @@ def send_single_email_view(request: WSGIRequest) -> HttpResponse: @require_POST @htmx_only("emails:dashboard") @feature_flag_check("areUserEmailsAllowed", status=True, api=True, htmx=True) -def send_bulk_email_view(request: WSGIRequest) -> HttpResponse: +def send_bulk_email_view(request: HtmxHttpRequest) -> HttpResponse: email_count = len(request.POST.getlist("emails")) - 1 check_usage = quota_usage_check_under(request, "emails-single-count", add=email_count, api=True, htmx=True) @@ -64,10 +65,10 @@ def send_bulk_email_view(request: WSGIRequest) -> HttpResponse: return _send_bulk_email_view(request) -def _send_bulk_email_view(request: WSGIRequest) -> HttpResponse: +def _send_bulk_email_view(request: HtmxHttpRequest) -> HttpResponse: emails: list[str] = request.POST.getlist("emails") - subject: str = request.POST.get("subject") - message: str = request.POST.get("content") + subject: str = request.POST.get("subject", "") + message: str = request.POST.get("content", "") if request.user.logged_in_as_team: clients = Client.objects.filter(organization=request.user.logged_in_as_team, email__in=emails) @@ -114,11 +115,11 @@ def _send_bulk_email_view(request: WSGIRequest) -> HttpResponse: return render(request, "base/toast.html") EMAIL_RESPONSES: Iterator[tuple[BulkEmailEmailItem, BulkEmailEntryResultTypeDef]] = zip( - EMAIL_DATA.email_list, EMAIL_SENT.response.get("BulkEmailEntryResults") + EMAIL_DATA.email_list, EMAIL_SENT.response.get("BulkEmailEntryResults") # type: ignore[arg-type] ) if request.user.logged_in_as_team: - SEND_STATUS_OBJECTS: QuerySet[EmailSendStatus] = EmailSendStatus.objects.bulk_create( + SEND_STATUS_OBJECTS: list[EmailSendStatus] = EmailSendStatus.objects.bulk_create( [ EmailSendStatus( organization=request.user.logged_in_as_team, @@ -131,7 +132,7 @@ def _send_bulk_email_view(request: WSGIRequest) -> HttpResponse: ] ) else: - SEND_STATUS_OBJECTS: QuerySet[EmailSendStatus] = EmailSendStatus.objects.bulk_create( + SEND_STATUS_OBJECTS = EmailSendStatus.objects.bulk_create( [ EmailSendStatus( user=request.user, @@ -162,10 +163,10 @@ def _send_bulk_email_view(request: WSGIRequest) -> HttpResponse: return render(request, "base/toast.html") -def _send_single_email_view(request: WSGIRequest) -> HttpResponse: +def _send_single_email_view(request: HtmxHttpRequest) -> HttpResponse: email: str = str(request.POST.get("email", "")).strip() - subject: str = request.POST.get("subject") - message: str = request.POST.get("content") + subject: str = request.POST.get("subject", "") + message: str = request.POST.get("content", "") if request.user.logged_in_as_team: client = Client.objects.filter(organization=request.user.logged_in_as_team, email=email).first() @@ -197,7 +198,11 @@ def _send_single_email_view(request: WSGIRequest) -> HttpResponse: EMAIL_SENT = send_email(data=EMAIL_DATA) - status_object = EmailSendStatus(sent_by=request.user, recipient=email, aws_message_id=EMAIL_SENT.response.get("MessageId")) + aws_message_id = None + if EMAIL_SENT.response is not None: + aws_message_id = EMAIL_SENT.response.get("MessageId") + + status_object = EmailSendStatus(sent_by=request.user, recipient=email, aws_message_id=aws_message_id) if EMAIL_SENT.success: messages.success(request, f"Successfully emailed {email}.") @@ -247,7 +252,7 @@ def run_validations(): return None -def validate_bulk_quotas(*, request: WSGIRequest, emails: list) -> str | None: +def validate_bulk_quotas(*, request: HtmxHttpRequest, emails: list) -> str | None: email_count = len(emails) slugs = ["emails-bulk-count", "emails-bulk-max_sends"] @@ -331,7 +336,7 @@ def validate_email_subject(subject: str) -> str | None: return None -def validate_email_content(message: str, request: WSGIRequest) -> str | None: +def validate_email_content(message: str, request: HtmxHttpRequest) -> str | None: min_count = 64 max_count = QuotaLimit.objects.get(slug="emails-email_character_count").get_quota_limit(user=request.user) diff --git a/backend/api/emails/status.py b/backend/api/emails/status.py index f634f312..4b4bf3c7 100644 --- a/backend/api/emails/status.py +++ b/backend/api/emails/status.py @@ -2,7 +2,7 @@ from typing import TypedDict from django.contrib import messages -from django.http import HttpRequest, HttpResponse +from django.http import HttpResponse from django.shortcuts import render from django.views.decorators.http import require_POST from django_ratelimit.core import is_ratelimited @@ -10,13 +10,14 @@ from backend.decorators import htmx_only, feature_flag_check from backend.models import EmailSendStatus +from backend.types.htmx import HtmxHttpRequest from settings.helpers import EMAIL_CLIENT @require_POST @htmx_only("emails:dashboard") @feature_flag_check("areUserEmailsAllowed", status=True, api=True, htmx=True) -def get_status_view(request: HttpRequest, status_id: str) -> HttpResponse: +def get_status_view(request: HtmxHttpRequest, status_id: str) -> HttpResponse: try: if request.user.logged_in_as_team: EMAIL_STATUS = EmailSendStatus.objects.get(organization=request.user.logged_in_as_team, id=status_id) @@ -26,7 +27,7 @@ def get_status_view(request: HttpRequest, status_id: str) -> HttpResponse: messages.error(request, "Status not found") return render(request, "base/toast.html") - message_insight = get_message_insights(message_id=EMAIL_STATUS.aws_message_id) + message_insight = get_message_insights(message_id=EMAIL_STATUS.aws_message_id) # type: ignore[arg-type] if isinstance(message_insight, str): messages.error(request, message_insight) @@ -45,7 +46,7 @@ def get_status_view(request: HttpRequest, status_id: str) -> HttpResponse: @require_POST @htmx_only("emails:dashboard") @feature_flag_check("areUserEmailsAllowed", status=True, api=True, htmx=True) -def refresh_all_statuses_view(request: HttpRequest) -> HttpResponse: +def refresh_all_statuses_view(request: HtmxHttpRequest) -> HttpResponse: if is_ratelimited(request, group="email-refresh_all_statuses", key="user", rate="5/10m", increment=True) or is_ratelimited( request, group="email-refresh_all_statuses", key="user", rate="1/m", increment=True ): @@ -57,7 +58,12 @@ def refresh_all_statuses_view(request: HttpRequest) -> HttpResponse: ALL_STATUSES = EmailSendStatus.objects.filter(user=request.user) for status in ALL_STATUSES: - response = get_message_insights(message_id=status.aws_message_id) + response = get_message_insights(message_id=status.aws_message_id) # type: ignore[arg-type] + + if isinstance(response, str): + messages.error(request, response) + continue + important_info = get_important_info_from_response(response) status.status = important_info["status"] @@ -66,9 +72,9 @@ def refresh_all_statuses_view(request: HttpRequest) -> HttpResponse: ALL_STATUSES.bulk_update(ALL_STATUSES, fields=["status", "updated_status_at", "updated_at"]) messages.success(request, "All statuses have been refreshed") - response = HttpResponse(status=200) - response["HX-Refresh"] = "true" - return response + http_response = HttpResponse(status=200) + http_response["HX-Refresh"] = "true" + return http_response class ImportantInfo(TypedDict): diff --git a/backend/api/invoices/create/services/add.py b/backend/api/invoices/create/services/add.py index 8ec5129c..f2f9f78b 100644 --- a/backend/api/invoices/create/services/add.py +++ b/backend/api/invoices/create/services/add.py @@ -1,14 +1,15 @@ -from django.http import HttpRequest, JsonResponse +from django.http import JsonResponse from django.shortcuts import render from django.views.decorators.http import require_http_methods from backend.models import InvoiceProduct +from backend.types.htmx import HtmxHttpRequest @require_http_methods(["POST"]) -def add_service(request: HttpRequest): - context = {} - existing_service = request.POST.get("existing_service") +def add_service(request: HtmxHttpRequest): + context: dict = {} + existing_service = request.POST.get("existing_service", "") try: existing_service_obj = InvoiceProduct.objects.get(user=request.user, id=existing_service) @@ -22,10 +23,10 @@ def add_service(request: HttpRequest): list_of_current_rows = [row for row in zip(list_hours, list_service_name, list_service_description, list_price_per_hour)] if not existing_service: - hours = int(request.POST.get("post_hours")) + hours = int(request.POST.get("post_hours", "")) service_name = request.POST.get("post_service_name") service_description = request.POST.get("post_service_description") - price_per_hour = int(request.POST.get("post_rate")) + price_per_hour = int(request.POST.get("post_rate", "")) if not hours: return JsonResponse( @@ -55,7 +56,7 @@ def add_service(request: HttpRequest): } ) - if existing_service and existing_service_obj: + if existing_service and existing_service_obj and existing_service_obj.rate is not None: context["rows"].append( { "hours": existing_service_obj.quantity, diff --git a/backend/api/invoices/create/set_destination.py b/backend/api/invoices/create/set_destination.py index ca2b5a1a..31c5feec 100644 --- a/backend/api/invoices/create/set_destination.py +++ b/backend/api/invoices/create/set_destination.py @@ -1,18 +1,18 @@ from django.contrib import messages -from django.http import HttpRequest from django.shortcuts import render from django.views.decorators.http import require_http_methods from backend.models import Client +from backend.types.htmx import HtmxHttpRequest to_get = ["name", "address", "city", "country", "company", "is_representative"] @require_http_methods(["POST"]) -def set_destination_to(request: HttpRequest): - context = {"swapping": True} +def set_destination_to(request: HtmxHttpRequest): + context: dict = {"swapping": True} - context.update({key: request.POST.get(key) for key in to_get}) + context.update({key: request.POST.get(key, "") for key in to_get}) use_existing = True if request.POST.get("use_existing") == "true" else False selected_client = request.POST.get("selected_client") if use_existing else None @@ -28,9 +28,9 @@ def set_destination_to(request: HttpRequest): @require_http_methods(["POST"]) -def set_destination_from(request: HttpRequest): - context = {"swapping": True} +def set_destination_from(request: HtmxHttpRequest): + context: dict = {"swapping": True} - context.update({key: request.POST.get(key) for key in to_get}) + context.update({key: request.POST.get(key, "") for key in to_get}) return render(request, "pages/invoices/create/_from_destination.html", context) diff --git a/backend/api/invoices/delete.py b/backend/api/invoices/delete.py index ffc1d765..55dca70b 100644 --- a/backend/api/invoices/delete.py +++ b/backend/api/invoices/delete.py @@ -1,21 +1,22 @@ from django.contrib import messages -from django.http import HttpRequest, JsonResponse, QueryDict, HttpResponse, HttpResponseRedirect +from django.http import JsonResponse, QueryDict, HttpResponse, HttpResponseRedirect from django.shortcuts import render from django.urls import resolve, reverse from django.urls.exceptions import Resolver404 from django.views.decorators.http import require_http_methods + from backend.models import Invoice, QuotaLimit +from backend.types.htmx import HtmxHttpRequest @require_http_methods(["DELETE"]) -def delete_invoice(request: HttpRequest): +def delete_invoice(request: HtmxHttpRequest): delete_items = QueryDict(request.body) - invoice = delete_items.get("invoice") redirect = delete_items.get("redirect", None) try: - invoice = Invoice.objects.get(id=invoice) + invoice = Invoice.objects.get(id=delete_items.get("invoice", "")) except Invoice.DoesNotExist: return JsonResponse({"message": "Invoice not found"}, status=404) diff --git a/backend/api/invoices/edit.py b/backend/api/invoices/edit.py index 6f80fdec..c278aae2 100644 --- a/backend/api/invoices/edit.py +++ b/backend/api/invoices/edit.py @@ -7,12 +7,13 @@ from django.views.decorators.http import require_http_methods, require_POST from backend.models import Invoice +from backend.types.htmx import HtmxHttpRequest @require_http_methods(["POST"]) -def edit_invoice(request: HttpRequest): +def edit_invoice(request: HtmxHttpRequest): try: - invoice = Invoice.objects.get(id=request.POST.get("invoice_id")) + invoice = Invoice.objects.get(id=request.POST.get("invoice_id", "")) except Invoice.DoesNotExist: return JsonResponse({"message": "Invoice not found"}, status=404) @@ -28,7 +29,7 @@ def edit_invoice(request: HttpRequest): ) attributes_to_updates = { - "date_due": datetime.strptime(request.POST.get("date_due"), "%Y-%m-%d").date(), + "date_due": datetime.strptime(request.POST.get("date_due", ""), "%Y-%m-%d").date(), "date_issued": request.POST.get("date_issued"), "client_name": request.POST.get("to_name"), "client_company": request.POST.get("to_company"), @@ -64,7 +65,7 @@ def edit_invoice(request: HttpRequest): @require_POST -def change_status(request: HttpRequest, invoice_id: int, status: str) -> HttpResponse: +def change_status(request: HtmxHttpRequest, invoice_id: int, status: str) -> HttpResponse: status = status.lower() if status else "" if not request.htmx: @@ -101,10 +102,10 @@ def change_status(request: HttpRequest, invoice_id: int, status: str) -> HttpRes @require_POST -def edit_discount(request: HttpRequest, invoice_id: str): +def edit_discount(request: HtmxHttpRequest, invoice_id: str): discount_type = "percentage" if request.POST.get("discount_type") == "on" else "amount" - discount_amount_str: str = request.POST.get("discount_amount") - percentage_amount_str: str = request.POST.get("percentage_amount") + discount_amount_str: str = request.POST.get("discount_amount", "") + percentage_amount_str: str = request.POST.get("percentage_amount", "") if not request.htmx: return redirect("invoices:dashboard") @@ -148,7 +149,7 @@ def return_message(request: HttpRequest, message: str, success: bool = True) -> return render(request, "base/toasts.html") -def send_message(request: HttpRequest, message: str, success: bool = False) -> NoReturn: +def send_message(request: HttpRequest, message: str, success: bool = False) -> None: if success: messages.success(request, message) else: diff --git a/backend/api/invoices/fetch.py b/backend/api/invoices/fetch.py index 0b5c9734..f35f0e25 100644 --- a/backend/api/invoices/fetch.py +++ b/backend/api/invoices/fetch.py @@ -10,25 +10,25 @@ Sum, Prefetch, ) -from django.http import HttpRequest from django.shortcuts import render, redirect from django.utils import timezone from django.views.decorators.http import require_http_methods from backend.models import Invoice, InvoiceItem +from backend.types.htmx import HtmxHttpRequest @require_http_methods(["GET"]) -def fetch_all_invoices(request: HttpRequest): +def fetch_all_invoices(request: HtmxHttpRequest): # Redirect if not an HTMX request if not request.htmx: return redirect("invoices:dashboard") - context = {} + context: dict = {} # Get filter and sort parameters from the request sort_by = request.GET.get("sort") - sort_direction = request.GET.get("sort_direction") + sort_direction = request.GET.get("sort_direction") if request.GET.get("sort_direction") else "" action_filter_type = request.GET.get("filter_type") action_filter_by = request.GET.get("filter") @@ -98,7 +98,7 @@ def fetch_all_invoices(request: HttpRequest): # Construct filter condition dynamically based on filter_type if "+" in filter_by: numeric_part = float(filter_by.split("+")[0]) - filter_condition = {f"{filter_type}__gte": numeric_part} + filter_condition: dict[str, str | float] = {f"{filter_type}__gte": numeric_part} else: filter_condition = {f"{filter_type}": filter_by} or_conditions_filter |= Q(**filter_condition) @@ -138,7 +138,7 @@ def fetch_all_invoices(request: HttpRequest): elif sort_by in all_sort_options: # True is for reverse order # first time set direction is none - if sort_direction.lower() == "true" or sort_direction == "": + if sort_direction is not None and sort_direction.lower() == "true" or sort_direction == "": context["sort"] = f"-{sort_by}" context["sort_direction"] = False invoices = invoices.order_by(f"-{sort_by}") diff --git a/backend/api/invoices/manage.py b/backend/api/invoices/manage.py index baf43d65..428c260e 100644 --- a/backend/api/invoices/manage.py +++ b/backend/api/invoices/manage.py @@ -11,7 +11,7 @@ from django.views.decorators.http import require_http_methods from backend.models import Invoice -from backend.models import UserSettings +from backend.types.htmx import HtmxHttpRequest class PreviewContext(TypedDict): @@ -33,7 +33,7 @@ class ErrorResponse: @require_http_methods(["GET"]) -def tab_preview_invoice(request: HttpRequest, invoice_id): +def tab_preview_invoice(request: HtmxHttpRequest, invoice_id): # Redirect if not an HTMX request if not request.htmx: return redirect("invoices dashboard") # Maybe should be 404? @@ -48,8 +48,8 @@ def tab_preview_invoice(request: HttpRequest, invoice_id): return render(request, "base/toasts.html") -def preview_invoice(request: HttpRequest, invoice_id) -> SuccessResponse | ErrorResponse: - context = {"type": "preview"} +def preview_invoice(request: HtmxHttpRequest, invoice_id) -> SuccessResponse | ErrorResponse: + context: dict[str, str | Invoice] = {"type": "preview"} try: invoice = Invoice.objects.prefetch_related("items").get(id=invoice_id) diff --git a/backend/api/invoices/reminders/create.py b/backend/api/invoices/reminders/create.py index 07b1538c..d3e9e99a 100644 --- a/backend/api/invoices/reminders/create.py +++ b/backend/api/invoices/reminders/create.py @@ -1,3 +1,5 @@ +from datetime import timedelta + from django.contrib import messages from django.core.handlers.wsgi import WSGIRequest from django.http import HttpResponse @@ -15,12 +17,12 @@ def get_datetime_from_reminder(reminder: InvoiceReminder) -> str: if reminder.reminder_type == "on_overdue": return reminder.invoice.date_due.strftime("%Y-%m-%dT%H:%M") - if reminder.reminder_type == "before_due": + if reminder.reminder_type == "before_due" and reminder.days: days = 0 - reminder.days else: - days = reminder.days + days = 0 - date = (timezone.now() + timezone.timedelta(days=days)).replace(hour=12, minute=0, second=0, microsecond=0).strftime("%Y-%m-%dT%H:%M") + date = (timezone.now() + timedelta(days=days)).replace(hour=12, minute=0, second=0, microsecond=0).strftime("%Y-%m-%dT%H:%M") return date @@ -34,9 +36,9 @@ def create_reminder_view(request: HtmxHttpRequest) -> HttpResponse: return check_usage # Extract POST data - invoice_id = request.POST.get("invoice_id") or request.POST.get("invoice") + invoice_id = request.POST.get("invoice_id", "") or request.POST.get("invoice", "") reminder_type = request.POST.get("reminder_type") - days = request.POST.get("days", "none") + days: str | int = request.POST.get("days", "none") # Check if invoice exists try: diff --git a/backend/api/invoices/reminders/delete.py b/backend/api/invoices/reminders/delete.py index 6718dcbf..3f45e8d5 100644 --- a/backend/api/invoices/reminders/delete.py +++ b/backend/api/invoices/reminders/delete.py @@ -39,7 +39,7 @@ def cancel_reminder_view(request: HtmxHttpRequest, reminder_id: str): return render(request, "pages/invoices/schedules/reminders/_table_row.html", {"reminder": reminder}) else: reminder.set_status(original_status) - messages.error(request, f"Failed to delete schedule: {delete_status['error']}") + messages.error(request, f"Failed to delete schedule: {delete_status.message}") return render(request, "base/toasts.html") reminder.set_status("cancelled") diff --git a/backend/api/invoices/reminders/fetch.py b/backend/api/invoices/reminders/fetch.py index be909c1e..e59021e3 100644 --- a/backend/api/invoices/reminders/fetch.py +++ b/backend/api/invoices/reminders/fetch.py @@ -1,17 +1,17 @@ from django.contrib import messages from django.db.models import Q -from django.http import HttpRequest from django.shortcuts import render from django.views.decorators.http import require_GET from django_ratelimit.core import is_ratelimited from backend.decorators import feature_flag_check from backend.models import Invoice +from backend.types.htmx import HtmxHttpRequest @require_GET @feature_flag_check("areInvoiceRemindersEnabled", True, api=True, htmx=True) -def fetch_reminders(request: HttpRequest, invoice_id: str): +def fetch_reminders(request: HtmxHttpRequest, invoice_id: str): ratelimit = is_ratelimited(request, group="fetch_reminders", key="user", rate="20/30s", increment=True) if ratelimit: messages.error(request, "Too many requests") @@ -27,7 +27,7 @@ def fetch_reminders(request: HttpRequest, invoice_id: str): messages.error(request, "You do not have permission to view this invoice") return render(request, "base/toasts.html") - context = {} + context: dict = {} reminders = invoice.invoice_reminders.order_by("reminder_type").only("id", "days", "reminder_type") diff --git a/backend/api/invoices/schedule.py b/backend/api/invoices/schedule.py index 92451c36..6f8d70b2 100644 --- a/backend/api/invoices/schedule.py +++ b/backend/api/invoices/schedule.py @@ -8,6 +8,7 @@ from django.views.decorators.http import require_http_methods, require_POST, require_GET from django_ratelimit.core import is_ratelimited from mypy_boto3_iam.type_defs import GetRoleResponseTypeDef +from mypy_boto3_scheduler.type_defs import ScheduleSummaryTypeDef from backend.decorators import feature_flag_check from backend.models import Invoice, AuditLog, APIKey, InvoiceOnetimeSchedule, InvoiceURL, QuotaUsage @@ -60,7 +61,7 @@ def create_schedule(request: HtmxHttpRequest): def create_ots(request: HtmxHttpRequest) -> HttpResponse: - invoice_id = request.POST.get("invoice_id") or request.POST.get("invoice") + invoice_id = request.POST.get("invoice_id", "") or request.POST.get("invoice", "") try: invoice = Invoice.objects.get(id=invoice_id) @@ -78,7 +79,7 @@ def create_ots(request: HtmxHttpRequest) -> HttpResponse: schedule = create_onetime_schedule( CreateOnetimeScheduleInputData( - invoice=invoice, option=1, datetime=request.POST.get("date_time"), email_type=request.POST.get("email_type") + invoice=invoice, option=1, datetime=request.POST.get("date_time"), email_type=request.POST.get("email_type") # type: ignore[arg-type] ) ) @@ -99,10 +100,10 @@ def get_execution_role() -> str | None: """ iam_client = get_iam_client() - response = iam_client.list_roles(PathPrefix=f"/{AWS_TAGS_APP_NAME}-scheduled-invoices/", MaxItems=1) + response: GetRoleResponseTypeDef = iam_client.list_roles(PathPrefix=f"/{AWS_TAGS_APP_NAME}-scheduled-invoices/", MaxItems=1) try: - response: GetRoleResponseTypeDef = iam_client.get_role(RoleName=f"{AWS_TAGS_APP_NAME}-invoicing-scheduler") + response = iam_client.get_role(RoleName=f"{AWS_TAGS_APP_NAME}-invoicing-scheduler") except (iam_client.exceptions.NoSuchEntityException, iam_client.exceptions.ServiceFailureException): return None @@ -199,15 +200,15 @@ def fetch_onetime_schedules(request: HtmxHttpRequest, invoice_id: str): messages.error(request, "Invoice not found") return render(request, "base/toasts.html") - if not invoice.user.logged_in_as_team and invoice.user != request.user: + if invoice.user and not invoice.user.logged_in_as_team and invoice.user != request.user: messages.error(request, "You do not have permission to view this invoice") return render(request, "base/toasts.html") - if invoice.user.logged_in_as_team and invoice.organization != request.user.logged_in_as_team: + if invoice.user and invoice.user.logged_in_as_team and invoice.organization != request.user.logged_in_as_team: messages.error(request, "You do not have permission to view this invoice") return render(request, "base/toasts.html") - context: dict[str, list[str] | dict] = {} + context: dict = {} schedules = invoice.onetime_invoice_schedules.order_by("due").only("id", "due", "status") @@ -268,7 +269,7 @@ def fetch_onetime_schedules(request: HtmxHttpRequest, invoice_id: str): messages.error(request, aws_schedules.message) else: # convert list of dictionaries to dictionary with key of ARN - aws_schedules = {schedule["Arn"]: schedule for schedule in aws_schedules.schedules} + aws_schedules_dict = {schedule["Arn"]: schedule for schedule in aws_schedules.schedules} for schedule in schedules: arn = schedule.stored_schedule_arn @@ -279,7 +280,7 @@ def fetch_onetime_schedules(request: HtmxHttpRequest, invoice_id: str): schedule.save() continue - if arn in aws_schedules: + if arn in aws_schedules_dict: if schedule.status != "pending" and schedule.status != "cancelled": schedule.status = "pending" schedule.save() diff --git a/backend/api/products/create.py b/backend/api/products/create.py index 338bec9c..127ca242 100644 --- a/backend/api/products/create.py +++ b/backend/api/products/create.py @@ -1,18 +1,16 @@ from django.contrib import messages -from django.db.models import Q -from django.http import HttpRequest -from django.shortcuts import render from backend.api.products.fetch import fetch_products from backend.models import InvoiceProduct +from backend.types.htmx import HtmxHttpRequest -def create_product(request: HttpRequest): +def create_product(request: HtmxHttpRequest): try: - quantity = int(request.POST.get("post_quantity")) - service_name = request.POST.get("post_service_name") - service_description = request.POST.get("post_service_description") - rate = int(request.POST.get("post_rate")) + quantity = int(request.POST.get("post_quantity", "")) + service_name = request.POST.get("post_service_name", "") + service_description = request.POST.get("post_service_description", "") + rate = int(request.POST.get("post_rate", "")) product = InvoiceProduct.objects.create( user=request.user, diff --git a/backend/api/products/fetch.py b/backend/api/products/fetch.py index 7e01e638..0e455476 100644 --- a/backend/api/products/fetch.py +++ b/backend/api/products/fetch.py @@ -1,12 +1,12 @@ -from django.db.models import Q -from django.http import HttpRequest +from django.db.models import Q, QuerySet from django.shortcuts import render from backend.models import InvoiceProduct +from backend.types.htmx import HtmxHttpRequest -def fetch_products(request: HttpRequest): - results = [] +def fetch_products(request: HtmxHttpRequest): + results: QuerySet search_text = request.GET.get("search_existing_service") if search_text: results = ( diff --git a/backend/api/quotas/fetch.py b/backend/api/quotas/fetch.py index 69aa93d8..a86e8597 100644 --- a/backend/api/quotas/fetch.py +++ b/backend/api/quotas/fetch.py @@ -1,11 +1,11 @@ from django.db.models import Q -from django.http import HttpRequest from django.shortcuts import render, redirect from backend.models import QuotaLimit +from backend.types.htmx import HtmxHttpRequest -def fetch_all_quotas(request: HttpRequest, group: str): +def fetch_all_quotas(request: HtmxHttpRequest, group: str): context = {} if not request.htmx: return redirect("quotas") diff --git a/backend/api/quotas/requests.py b/backend/api/quotas/requests.py index acf4602d..45b0f503 100644 --- a/backend/api/quotas/requests.py +++ b/backend/api/quotas/requests.py @@ -2,21 +2,22 @@ from typing import Union from django.contrib import messages -from django.http import HttpRequest, HttpResponse +from django.http import HttpResponse from django.shortcuts import redirect, render from django.views.decorators.http import require_http_methods from backend.decorators import superuser_only from backend.models import QuotaIncreaseRequest, QuotaLimit, QuotaUsage, QuotaOverrides +from backend.types.htmx import HtmxHttpRequest from backend.utils.quota_limit_ops import quota_usage_check_under -def submit_request(request: HttpRequest, slug) -> HttpResponse: +def submit_request(request: HtmxHttpRequest, slug) -> HttpResponse: if not request.htmx: return redirect("quotas") - new_value = request.POST.get("new_value") - reason = request.POST.get("reason") + new_value = request.POST.get("new_value", "") + reason = request.POST.get("reason", "") try: quota_limit = QuotaLimit.objects.get(slug=slug) @@ -57,7 +58,7 @@ class Error: message: str -def error(request: HttpRequest, message: str) -> HttpResponse: +def error(request: HtmxHttpRequest, message: str) -> HttpResponse: messages.error(request, message) return render(request, "partials/messages_list.html") @@ -81,7 +82,7 @@ def validate_request(new_value, reason, current) -> Union[bool, Error]: @superuser_only @require_http_methods(["DELETE", "POST"]) -def approve_request(request: HttpRequest, request_id) -> HttpResponse: +def approve_request(request: HtmxHttpRequest, request_id) -> HttpResponse: if not request.htmx: return redirect("quotas") try: @@ -117,7 +118,7 @@ def approve_request(request: HttpRequest, request_id) -> HttpResponse: @superuser_only @require_http_methods(["DELETE", "POST"]) -def decline_request(request: HttpRequest, request_id) -> HttpResponse: +def decline_request(request: HtmxHttpRequest, request_id) -> HttpResponse: if not request.htmx: return redirect("quotas") try: diff --git a/backend/api/receipts/fetch.py b/backend/api/receipts/fetch.py index 0c8fb943..8cd3edd7 100644 --- a/backend/api/receipts/fetch.py +++ b/backend/api/receipts/fetch.py @@ -1,4 +1,4 @@ -from django.db.models import Q +from django.db.models import Q, QuerySet from django.shortcuts import render, redirect from backend.models import Receipt, User @@ -6,7 +6,7 @@ def fetch_all_receipts(request: HtmxHttpRequest): - context = {} + context: dict[str, QuerySet | list[str] | dict[str, list[str]]] = {} if not request.htmx: return redirect("receipts dashboard") diff --git a/backend/api/receipts/new.py b/backend/api/receipts/new.py index a5a65fe8..a94b9523 100644 --- a/backend/api/receipts/new.py +++ b/backend/api/receipts/new.py @@ -1,17 +1,18 @@ from django.contrib import messages from django.contrib.auth.decorators import login_required -from django.http import HttpRequest, HttpResponseBadRequest +from django.http import HttpResponseBadRequest from django.shortcuts import render, redirect from django.views.decorators.http import require_http_methods from backend.decorators import quota_usage_check from backend.models import Receipt, QuotaUsage +from backend.types.htmx import HtmxHttpRequest @require_http_methods(["POST"]) @quota_usage_check("receipts-count", api=True, htmx=True) @login_required -def receipt_create(request: HttpRequest): +def receipt_create(request: HtmxHttpRequest): if not request.htmx: return redirect("receipts dashboard") file = request.FILES.get("receipt_image") # InMemoryUploadedFile @@ -25,7 +26,7 @@ def receipt_create(request: HttpRequest): messages.error(request, "No image found") return HttpResponseBadRequest("No image found", status=400) - name = file.name.split(".")[0] if not name else name + name = file.name.split(".")[0] if not name and file.name else name if not name: messages.error(request, "No name provided, or image doesn't contain a valid name.") @@ -34,7 +35,7 @@ def receipt_create(request: HttpRequest): if not date: date = None - receipt_data = { + receipt_data: dict = { "name": name, "image": file, "date": date, diff --git a/backend/api/teams/create.py b/backend/api/teams/create.py index 9942dca1..7d798300 100644 --- a/backend/api/teams/create.py +++ b/backend/api/teams/create.py @@ -1,15 +1,19 @@ -from django.http import HttpRequest from django.views.decorators.http import require_POST from backend.decorators import * from backend.models import Team, QuotaUsage +from backend.types.htmx import HtmxHttpRequest @require_POST @quota_usage_check("teams-count", api=True, htmx=True) -def create_team(request: HttpRequest): +def create_team(request: HtmxHttpRequest): name = request.POST.get("name") + if not name: + messages.error(request, "A team name field must be filled.") + return render(request, "partials/messages_list.html") + if Team.objects.filter(name=name).exists(): messages.error(request, "A team with this name already exists.") return render(request, "partials/messages_list.html") diff --git a/backend/api/teams/invites.py b/backend/api/teams/invites.py index e5b40d73..65731147 100644 --- a/backend/api/teams/invites.py +++ b/backend/api/teams/invites.py @@ -1,29 +1,26 @@ -from django.http import HttpRequest - from backend.decorators import * from backend.models import Notification, Team, TeamInvitation, User +from backend.types.htmx import HtmxHttpRequest -def delete_notification(user: User, code: TeamInvitation.code): - Notification.objects.filter( +def delete_notification(user: User, code: TeamInvitation): + notification = Notification.objects.filter( user=user, message="New Team Invite", action="modal", action_value="accept_invite", extra_type="accept_invite_with_code", extra_value=code, - ).first().delete() + ).first() + + if notification: + notification.delete() def check_team_invitation_is_valid(request, invitation: TeamInvitation, code=None): valid: bool = True - if not invitation: - messages.error(request, "Invalid Invite Code") - # Force break early to avoid "no invitation" on invitation.code - delete_notification(request.user, code) - return False - if not invitation.is_active: + if not invitation.is_active(): valid = False messages.error(request, "Invitation has expired") @@ -43,14 +40,14 @@ def check_team_invitation_is_valid(request, invitation: TeamInvitation, code=Non return True -def send_user_team_invite(request: HttpRequest): +def send_user_team_invite(request: HtmxHttpRequest): user_email = request.POST.get("email") - team_id = request.POST.get("team_id") + team_id = request.POST.get("team_id", "") team = Team.objects.filter(leader=request.user, id=team_id).first() - def return_error_notif(request: HttpRequest, message: str, autohide=None): + def return_error_notif(request: HtmxHttpRequest, message: str, autohide=None): messages.error(request, message) - context = {"autohide": False} if autohide == False else {} + context = {"autohide": False} if autohide is False else {} resp = render(request, "partials/messages_list.html", context=context, status=200) resp["HX-Trigger-After-Swap"] = "invite_user_error" return resp @@ -61,7 +58,7 @@ def return_error_notif(request: HttpRequest, message: str, autohide=None): if not team: return return_error_notif(request, "You are not the leader of this team") - user: User = User.objects.filter(email=user_email).first() + user: User | None = User.objects.filter(email=user_email).first() if not user: return return_error_notif(request, "User not found") @@ -115,8 +112,14 @@ def return_error_notif(request: HttpRequest, message: str, autohide=None): return response -def accept_team_invite(request: HttpRequest, code): - invitation: TeamInvitation = TeamInvitation.objects.filter(code=code).prefetch_related("team").first() +def accept_team_invite(request: HtmxHttpRequest, code): + invitation: TeamInvitation | None = TeamInvitation.objects.filter(code=code).prefetch_related("team").first() + + if not invitation: + messages.error(request, "Invalid Invite Code") + # Force break early to avoid "no invitation" on invitation.code + delete_notification(request.user, code) + return render(request, "partials/messages_list.html") if not check_team_invitation_is_valid(request, invitation, code): messages.error(request, "Invalid invite - Maybe it has expired?") @@ -162,10 +165,16 @@ def accept_team_invite(request: HttpRequest, code): # return render(request, "partials/messages_list.html") -def decline_team_invite(request: HttpRequest, code): - invitation: TeamInvitation = TeamInvitation.objects.filter(code=code).first() +def decline_team_invite(request: HtmxHttpRequest, code): + invitation: TeamInvitation | None = TeamInvitation.objects.filter(code=code).first() confirmation_text = request.POST.get("confirmation_text") + if not invitation: + messages.error(request, "Invalid Invite Code") + # Force break early to avoid "no invitation" on invitation.code + delete_notification(request.user, code) + return render(request, "partials/messages_list.html") + if not check_team_invitation_is_valid(request, invitation, code): return render(request, "partials/messages_list.html") diff --git a/backend/types/htmx.py b/backend/types/htmx.py index 29fca486..b6dc6a69 100644 --- a/backend/types/htmx.py +++ b/backend/types/htmx.py @@ -1,4 +1,5 @@ from django.contrib.auth.models import AnonymousUser +from django.core.handlers.wsgi import WSGIRequest from django.http import HttpRequest from django_htmx.middleware import HtmxDetails diff --git a/backend/utils/dataclasses.py b/backend/utils/dataclasses.py index a27b7534..cfb54119 100644 --- a/backend/utils/dataclasses.py +++ b/backend/utils/dataclasses.py @@ -4,7 +4,7 @@ T = TypeVar("T") -def extract_to_dataclass(request, class_type: [T], request_types: list[str], *args, **kwargs) -> [T]: +def extract_to_dataclass(request, class_type: T, request_types: list[str], *args, **kwargs) -> T: """ Turn kwargs from Key:Value and get request.POST.get(value) and set class.key = request.POST.get(value) @@ -27,7 +27,7 @@ def myview(request): except pydantic.ValidationError: pass """ - data = {} + data: dict = {} if "get" in request_types: if args: data |= {key: request.GET.get(key) for key in args} @@ -48,4 +48,8 @@ def myview(request): if kwargs: data |= {key: request.headers.get(value) for key, value in kwargs.items()} - return class_type(**data) + + if isinstance(class_type, type): + return class_type(**data) + else: + raise TypeError("class_type must be a class") diff --git a/infrastructure/aws/schedules/create_schedule.py b/infrastructure/aws/schedules/create_schedule.py index 26f426bb..f253eb63 100644 --- a/infrastructure/aws/schedules/create_schedule.py +++ b/infrastructure/aws/schedules/create_schedule.py @@ -5,6 +5,7 @@ import pytz from django.utils import timezone from rest_framework.reverse import reverse +from datetime import datetime from backend.models import Invoice, InvoiceOnetimeSchedule, AuditLog from infrastructure.aws.handler import get_event_bridge_scheduler @@ -51,7 +52,7 @@ def create_onetime_schedule(data: CreateOnetimeScheduleInputData) -> CreateOneti date_time = data.date + "T" + data.time - date_time_to_obj: timezone.datetime = timezone.datetime.strptime(date_time, "%Y-%m-%dT%H:%M") + date_time_to_obj: datetime = datetime.strptime(date_time, "%Y-%m-%dT%H:%M") date_time_to_obj = date_time_to_obj.astimezone(pytz.timezone("UTC")) if date_time_to_obj < timezone.now(): return ErrorResponse("Date time cannot be in the past") diff --git a/settings/helpers.py b/settings/helpers.py index 2173167b..7c5958b3 100644 --- a/settings/helpers.py +++ b/settings/helpers.py @@ -122,7 +122,9 @@ def send_email(data: SingleEmailInput) -> SingleEmailSuccessResponse | SingleEma response = EMAIL_CLIENT.send_email( FromEmailAddress=from_email_address, Destination={"ToAddresses": data.destination}, - Content={"Simple": {"Subject": {"Data": data.subject}, "Body": {"Text": {"Data": data.content}}}}, + Content={ + "Simple": {"Subject": {"Data": data.subject if data.subject else ""}, "Body": {"Text": {"Data": data.content}}} + }, ConfigurationSetName=data.ConfigurationSetName or "", ) return SingleEmailSuccessResponse(response) diff --git a/settings/local_settings.py b/settings/local_settings.py index 1081e1bb..466f2c3b 100644 --- a/settings/local_settings.py +++ b/settings/local_settings.py @@ -17,20 +17,19 @@ if DB_TYPE == "mysql" or DB_TYPE == "postgres": DATABASES: dict = { "default": { - "ENGINE": ( - "django.db.backends.postgresql_psycopg2" - if DB_TYPE == "mysql" - else "django.db.backends.postgresql" - ), + "ENGINE": ("django.db.backends.postgresql_psycopg2" if DB_TYPE == "mysql" else "django.db.backends.postgresql"), "NAME": os.environ.get("DATABASE_NAME") or "myfinances_development", "USER": os.environ.get("DATABASE_USER") or "root", "PASSWORD": os.environ.get("DATABASE_PASS") or "", "HOST": os.environ.get("DATABASE_HOST") or "localhost", - "PORT": os.environ.get("DATABASE_PORT") - or (3306 if DB_TYPE == "mysql" else 5432), - "OPTIONS": { - "sql_mode": "traditional", - } if DB_TYPE == "mysql" else {}, + "PORT": os.environ.get("DATABASE_PORT") or (3306 if DB_TYPE == "mysql" else 5432), + "OPTIONS": ( + { + "sql_mode": "traditional", + } + if DB_TYPE == "mysql" + else {} + ), } } @@ -47,6 +46,4 @@ ALLOWED_HOSTS: list[str | None] = ["localhost", "127.0.0.1"] -os.environ["OAUTHLIB_INSECURE_TRANSPORT"] = ( - "1" # THIS WILL ALLOW HTTP - NOT RECOMMENDED -) +os.environ["OAUTHLIB_INSECURE_TRANSPORT"] = "1" # THIS WILL ALLOW HTTP - NOT RECOMMENDED diff --git a/settings/prod_settings.py b/settings/prod_settings.py index 6f4ea91a..94e6487d 100644 --- a/settings/prod_settings.py +++ b/settings/prod_settings.py @@ -41,6 +41,6 @@ print(f"[BACKEND] Using {DB_TYPE} database: {os.environ.get('DATABASE_NAME')}") -ALLOWED_HOSTS: list[str | None] = [os.environ.get("URL")] + URL_LIST +ALLOWED_HOSTS: list[str | None] = [os.environ.get("URL")] + URL_LIST # type: ignore os.environ["OAUTHLIB_INSECURE_TRANSPORT"] = "0" # THIS WILL ALLOW HTTP IF IT'S SET TO 1 - NOT RECOMMENDED