Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Add basic typing information to module froide.foirequest. #492

Open
wants to merge 7 commits into
base: main
Choose a base branch
from
Open
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Jump to
Jump to file
Failed to load files.
Diff view
Diff view
122 changes: 82 additions & 40 deletions froide/foirequest/utils.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,20 +2,27 @@
import re
from dataclasses import dataclass
from datetime import timedelta
from typing import Iterator, List, Optional, Tuple, Union
from re import Pattern
from typing import Any, Dict, Iterator, List, Optional, Tuple, Type, Union

from django import forms
from django.conf import settings
from django.core.exceptions import ValidationError
from django.core.mail import mail_managers
from django.db.models.query import QuerySet
from django.template.loader import render_to_string
from django.urls import reverse
from django.utils import timezone
from django.utils.crypto import get_random_string
from django.utils.functional import SimpleLazyObject
from django.utils.safestring import SafeString
from django.utils.translation import gettext_lazy as _

import icalendar
import pytz

from froide.account.models import User
from froide.foirequest.models.request import FoiRequest
from froide.helper.date_utils import format_seconds
from froide.helper.email_utils import delete_mails_by_recipient
from froide.helper.tasks import search_instance_save
Expand All @@ -25,13 +32,18 @@
redact_subject,
redact_user_strings,
)
from froide.publicbody.models import PublicBody
from froide.publicbody.models import FoiLaw, PublicBody

MAX_ATTACHMENT_SIZE = settings.FROIDE_CONFIG["max_attachment_size"]
RECIPIENT_BLOCKLIST = settings.FROIDE_CONFIG.get("recipient_blocklist_regex", None)

ThrottleConfig = Optional[List[Tuple[int, int]]]
Throttle = Union[bool, Tuple[int, int]]

def throttle(qs, throttle_config, date_param="first_message"):

def throttle(
qs: QuerySet, throttle_config: ThrottleConfig, date_param: str = "first_message"
) -> Throttle:
if throttle_config is None:
return False

Expand All @@ -43,7 +55,11 @@ def throttle(qs, throttle_config, date_param="first_message"):
return False


def check_throttle(user, klass, extra_filters=None):
def check_throttle(
user: Union[User, SimpleLazyObject],
klass: Type[FoiRequest],
extra_filters: None = None,
) -> Optional[ValidationError]:
if not user.is_authenticated or user.trusted():
return

Expand Down Expand Up @@ -72,7 +88,9 @@ def get_foi_mail_domains() -> Union[List[str], Tuple[str]]:
return domains


def generate_secret_address(user, length=10):
def generate_secret_address(
user: Union[User, SimpleLazyObject], length: int = 10
) -> str:
allowed_chars = "abcdefghkmnprstuvwxyz2345689"
username = user.username.replace("_", ".")
secret = get_random_string(length=length, allowed_chars=allowed_chars)
Expand All @@ -89,14 +107,18 @@ def generate_secret_address(user, length=10):
return "%s.%s@%s" % (username, secret, FOI_EMAIL_DOMAIN)


AttachmentNames = Optional[List[Union[str, Any]]]
AttachmentMissing = Optional[List[Any]]


def construct_message_body(
foirequest,
text,
send_address=True,
attachment_names=None,
attachment_missing=None,
template="foirequest/emails/mail_with_userinfo.txt",
):
foirequest: FoiRequest,
text: str,
send_address: bool = True,
attachment_names: AttachmentNames = None,
attachment_missing: AttachmentMissing = None,
template: str = "foirequest/emails/mail_with_userinfo.txt",
) -> SafeString:
return render_to_string(
template,
{
Expand All @@ -109,7 +131,7 @@ def construct_message_body(
)


def build_secret_url_regexes(url_name):
def build_secret_url_regexes(url_name: str) -> Pattern:
obj_id = "0"
code = "deadbeef"
url = reverse(url_name, kwargs={"obj_id": obj_id, "code": code})
Expand All @@ -128,7 +150,7 @@ def build_secret_url_regexes(url_name):
SECRET_URL_REPLACEMENTS = {}


def get_secret_url_replacements():
def get_secret_url_replacements() -> Dict[Pattern, str]:
if SECRET_URL_REPLACEMENTS:
return SECRET_URL_REPLACEMENTS

Expand All @@ -141,7 +163,7 @@ def get_secret_url_replacements():
return SECRET_URL_REPLACEMENTS


def short_request_url(name, foirequest, message=None):
def short_request_url(name: str, foirequest: FoiRequest, message: None = None) -> str:
params = {"slug": foirequest.slug}
if message:
params["message_id"] = message.id
Expand All @@ -156,8 +178,11 @@ def short_request_url(name, foirequest, message=None):


def redact_plaintext_with_request(
plaintext, foirequest, redact_greeting=False, redact_closing=False
):
plaintext: Union[SafeString, str],
foirequest: FoiRequest,
redact_greeting: bool = False,
redact_closing: bool = False,
) -> str:
replacements = get_secret_url_replacements()
user_replacements = foirequest.user.get_redactions()
return redact_plaintext(
Expand All @@ -170,13 +195,13 @@ def redact_plaintext_with_request(


def construct_initial_message_body(
foirequest,
text="",
foilaw=None,
full_text=False,
send_address=True,
template="foirequest/emails/mail_with_userinfo.txt",
):
foirequest: FoiRequest,
text: str = "",
foilaw: Optional[FoiLaw] = None,
full_text: bool = False,
send_address: bool = True,
template: str = "foirequest/emails/mail_with_userinfo.txt",
) -> SafeString:
if full_text:
body = "{body}\n{name}".format(body=text, name=foirequest.user.get_full_name())
else:
Expand All @@ -201,17 +226,17 @@ def construct_initial_message_body(
)


def strip_subdomains(domain):
def strip_subdomains(domain: str) -> str:
return ".".join(domain.split(".")[-2:])


def get_host(email):
def get_host(email: str) -> str:
if email and "@" in email:
return email.rsplit("@", 1)[1].lower()
return None


def get_domain(email):
def get_domain(email: str) -> str:
host = get_host(email)
if host is None:
return None
Expand All @@ -227,7 +252,7 @@ def compare_publicbody_email(email, pb_info_list, transform=str.lower):


def get_publicbody_for_email(
email: str, foirequest, include_deferred=False
email: str, foirequest: FoiRequest, include_deferred: bool = False
) -> Optional[PublicBody]:
if not email:
return None
Expand Down Expand Up @@ -317,11 +342,11 @@ class PublicBodyEmailInfo:
publicbody: Optional[PublicBody] = None
label: Optional[str] = None

def get_label(self):
def get_label(self) -> str:
return self.label if self.label is not None else self.name


def get_info_for_email(foirequest, email):
def get_info_for_email(foirequest: FoiRequest, email: str) -> PublicBodyEmailInfo:
email = email.lower()
for email_info in get_emails_from_request(foirequest):
if email == email_info.email.lower():
Expand All @@ -330,7 +355,7 @@ def get_info_for_email(foirequest, email):


def get_emails_from_request(
foirequest, include_mediator=True
foirequest: FoiRequest, include_mediator: bool = True
) -> Iterator[PublicBodyEmailInfo]:
already = set()

Expand Down Expand Up @@ -419,7 +444,10 @@ def get_emails_from_request(
already.add(email)


def get_emails_from_message_headers(email_headers):
EmailHeaders = Dict[str, List[Any]]


def get_emails_from_message_headers(email_headers: EmailHeaders) -> None:
fields = ("to", "cc", "resent-to", "resent-cc")
for field in fields:
for addr in email_headers.get(field, []):
Expand All @@ -428,7 +456,10 @@ def get_emails_from_message_headers(email_headers):
yield email


def possible_reply_addresses(foirequest):
PossibleReplyAddresses = List[Tuple[str, str]]


def possible_reply_addresses(foirequest: FoiRequest) -> PossibleReplyAddresses:
FOI_MAIL_DOMAIN = get_foi_mail_domains()[0]
options = []
for email_info in get_emails_from_request(foirequest, include_mediator=False):
Expand All @@ -449,12 +480,18 @@ def possible_reply_addresses(foirequest):
return options


Generator = List[Tuple[str, bytes, str]]
MailAttachmentSizeCheckerIterator = Iterator[Tuple[str, bytes, str]]


class MailAttachmentSizeChecker:
def __init__(self, generator, max_size=MAX_ATTACHMENT_SIZE):
def __init__(
self, generator: Generator, max_size: int = MAX_ATTACHMENT_SIZE
) -> None:
self.generator = generator
self.max_size = max_size

def __iter__(self):
def __iter__(self) -> MailAttachmentSizeCheckerIterator:
sum_bytes = 0
self.non_send_files = []
self.send_files = []
Expand All @@ -467,7 +504,12 @@ def __iter__(self):
yield name, filebytes, mimetype


def merge_user(sender, old_user=None, new_user=None, **kwargs):
def merge_user(
sender: Type[User],
old_user: Optional[User] = None,
new_user: Optional[User] = None,
**kwargs
) -> None:
from froide.account.utils import move_ownership

from .models import (
Expand All @@ -492,7 +534,7 @@ def merge_user(sender, old_user=None, new_user=None, **kwargs):
update_foirequest_index(FoiRequest.objects.filter(user=new_user))


def cancel_user(sender, user=None, **kwargs):
def cancel_user(sender: Type[User], user: Optional[User] = None, **kwargs) -> None:
from .models import FoiRequest, RequestDraft

if user is None:
Expand Down Expand Up @@ -547,7 +589,7 @@ def rerun_message_redaction(foirequests):
message.save()


def permanently_anonymize_requests(foirequests):
def permanently_anonymize_requests(foirequests: QuerySet) -> None:
from .models import FoiAttachment

replacements = {
Expand Down Expand Up @@ -806,12 +848,12 @@ def export_user_data(user):
)


def update_foirequest_index(queryset):
def update_foirequest_index(queryset: QuerySet) -> None:
for foirequest_id in queryset.values_list("id", flat=True):
search_instance_save.delay("foirequest.foirequest", foirequest_id)


def select_foirequest_template(foirequest, base_template: str):
def select_foirequest_template(foirequest: FoiRequest, base_template: str) -> List[str]:
path, name = base_template.rsplit("/", 1)
templates = []
if foirequest.law and foirequest.law.law_type:
Expand Down