diff --git a/_modules/parsedmarc.html b/_modules/parsedmarc.html
index bc058ed..3bc62be 100644
--- a/_modules/parsedmarc.html
+++ b/_modules/parsedmarc.html
@@ -103,15 +103,10 @@
from csv import DictWriter
from datetime import date, datetime, timedelta, timezone, tzinfo
from io import BytesIO, StringIO
+from collections.abc import Callable, Sequence
from typing import (
Any,
BinaryIO,
- Callable,
- Dict,
- List,
- Optional,
- Sequence,
- Union,
cast,
)
@@ -262,7 +257,7 @@
return " (raised at {0}:{1})".format(last.filename, last.lineno)
-def _text(value: Any) -> Optional[str]:
+def _text(value: Any) -> str | None:
"""Unwrap a possibly-langAttrString value parsed by xmltodict.
RFC 9990 changed several aggregate-report elements (extra_contact_info,
@@ -286,7 +281,7 @@
begin: datetime,
end: datetime,
total_count: int,
-) -> List[Dict[str, Any]]:
+) -> list[dict[str, Any]]:
"""
Split the interval [begin, end) into daily buckets and distribute
`total_count` proportionally across those buckets.
@@ -348,7 +343,7 @@
if day_cursor > begin:
day_cursor -= timedelta(days=1)
- day_buckets: List[Dict[str, Any]] = []
+ day_buckets: list[dict[str, Any]] = []
while day_cursor < end:
day_start = day_cursor
@@ -380,12 +375,12 @@
# Then apply a "largest remainder" rounding strategy to ensure the sum
# equals exactly total_count.
- exact_values: List[float] = [
+ exact_values: list[float] = [
(b["seconds"] / interval_seconds) * total_count for b in day_buckets
]
- floor_values: List[int] = [int(x) for x in exact_values]
- fractional_parts: List[float] = [x - int(x) for x in exact_values]
+ floor_values: list[int] = [int(x) for x in exact_values]
+ fractional_parts: list[float] = [x - int(x) for x in exact_values]
# How many counts do we still need to distribute after flooring?
remainder = total_count - sum(floor_values)
@@ -405,7 +400,7 @@
final_counts[idx] += 1
# --- Step 3: Build the final per-day result list -------------------------
- results: List[Dict[str, Any]] = []
+ results: list[dict[str, Any]] = []
for bucket, count in zip(day_buckets, final_counts):
if count > 0:
results.append(
@@ -466,12 +461,12 @@
def _parse_report_record(
record: dict[str, Any],
*,
- ip_db_path: Optional[str] = None,
+ ip_db_path: str | None = None,
always_use_local_files: bool = False,
- reverse_dns_map_path: Optional[str] = None,
- reverse_dns_map_url: Optional[str] = None,
+ reverse_dns_map_path: str | None = None,
+ reverse_dns_map_url: str | None = None,
offline: bool = False,
- nameservers: Optional[list[str]] = None,
+ nameservers: list[str] | None = None,
dns_timeout: float = DEFAULT_DNS_TIMEOUT,
dns_retries: int = DEFAULT_DNS_MAX_RETRIES,
is_rfc_9990: bool = False,
@@ -730,7 +725,7 @@
[docs]
-
def parse_smtp_tls_report_json(report: Union[str, bytes]) -> SMTPTLSReport:
+
def parse_smtp_tls_report_json(report: str | bytes) -> SMTPTLSReport:
"""Parses and validates an SMTP TLS report"""
required_fields = [
"organization-name",
@@ -776,7 +771,7 @@
[docs]
def parsed_smtp_tls_reports_to_csv_rows(
-
reports: Union[SMTPTLSReport, list[SMTPTLSReport]],
+
reports: SMTPTLSReport | list[SMTPTLSReport],
) -> list[dict[str, Any]]:
"""Converts one oor more parsed SMTP TLS reports into a list of single
layer dict objects suitable for use in a CSV"""
@@ -819,7 +814,7 @@
[docs]
def parsed_smtp_tls_reports_to_csv(
-
reports: Union[SMTPTLSReport, list[SMTPTLSReport]],
+
reports: SMTPTLSReport | list[SMTPTLSReport],
) -> str:
"""
Converts one or more parsed SMTP TLS reports to flat CSV format, including
@@ -871,15 +866,15 @@
def parse_aggregate_report_xml(
xml: str,
*,
-
ip_db_path: Optional[str] = None,
+
ip_db_path: str | None = None,
always_use_local_files: bool = False,
-
reverse_dns_map_path: Optional[str] = None,
-
reverse_dns_map_url: Optional[str] = None,
+
reverse_dns_map_path: str | None = None,
+
reverse_dns_map_url: str | None = None,
offline: bool = False,
-
nameservers: Optional[list[str]] = None,
+
nameservers: list[str] | None = None,
timeout: float = DEFAULT_DNS_TIMEOUT,
retries: int = DEFAULT_DNS_MAX_RETRIES,
-
keep_alive: Optional[Callable] = None,
+
keep_alive: Callable | None = None,
normalize_timespan_threshold_hours: float = 24.0,
) -> AggregateReport:
"""Parses a DMARC XML report string and returns a consistent dict
@@ -916,7 +911,7 @@
# The final `is_rfc_9990` decision is made post-parse so that
# RFC 9990-only fields (np, testing, discovery_method, generator,
# human_result) can also vote it in.
-
xml_namespace: Optional[str] = None
+
xml_namespace: str | None = None
namespace_match = xml_namespace_regex.search(xml)
if namespace_match:
xml_namespace = namespace_match.group(1)
@@ -1172,7 +1167,7 @@
+
records: list[AggregateRecord]
[docs]
class EmailAddress(TypedDict):
-
display_name: Optional[str]
+
display_name: str | None
address: str
-
local: Optional[str]
-
domain: Optional[str]
+
local: str | None
+
domain: str | None
[docs]
class EmailAttachment(TypedDict, total=False):
-
filename: Optional[str]
-
mail_content_type: Optional[str]
-
sha256: Optional[str]
+
filename: str | None
+
mail_content_type: str | None
+
sha256: str | None
@@ -254,16 +255,16 @@
{
# This is a lightly-specified version of mailsuite/mailparser JSON.
# It focuses on the fields parsedmarc uses in failure report handling.
- "headers": Dict[str, Any],
- "subject": Optional[str],
- "filename_safe_subject": Optional[str],
- "date": Optional[str],
+ "headers": dict[str, Any],
+ "subject": str | None,
+ "filename_safe_subject": str | None,
+ "date": str | None,
"from": EmailAddress,
- "to": List[EmailAddress],
- "cc": List[EmailAddress],
- "bcc": List[EmailAddress],
- "attachments": List[EmailAttachment],
- "body": Optional[str],
+ "to": list[EmailAddress],
+ "cc": list[EmailAddress],
+ "bcc": list[EmailAddress],
+ "attachments": list[EmailAttachment],
+ "body": str | None,
"has_defects": bool,
"defects": Any,
"defects_categories": Any,
@@ -275,19 +276,19 @@
[docs]
class FailureReport(TypedDict):
-
feedback_type: Optional[str]
-
user_agent: Optional[str]
-
version: Optional[str]
-
original_envelope_id: Optional[str]
-
original_mail_from: Optional[str]
-
original_rcpt_to: Optional[str]
+
feedback_type: str | None
+
user_agent: str | None
+
version: str | None
+
original_envelope_id: str | None
+
original_mail_from: str | None
+
original_rcpt_to: str | None
arrival_date: str
arrival_date_utc: str
-
authentication_results: Optional[str]
-
delivery_result: Optional[str]
-
auth_failure: List[str]
-
authentication_mechanisms: List[str]
-
dkim_domain: Optional[str]
+
authentication_results: str | None
+
delivery_result: str | None
+
auth_failure: list[str]
+
authentication_mechanisms: list[str]
+
dkim_domain: str | None
reported_domain: str
sample_headers_only: bool
source: IPSourceInfo
@@ -334,9 +335,9 @@
[docs]
class SMTPTLSPolicy(SMTPTLSPolicySummary, total=False):
-
policy_strings: List[str]
-
mx_host_patterns: List[str]
-
failure_details: List[SMTPTLSFailureDetailsOptional]
+
policy_strings: list[str]
+
mx_host_patterns: list[str]
+
failure_details: list[SMTPTLSFailureDetailsOptional]
@@ -346,9 +347,9 @@
organization_name: str
begin_date: str
end_date: str
- contact_info: Union[str, List[str]]
+ contact_info: str | list[str]
report_id: str
- policies: List[SMTPTLSPolicy]
+ policies: list[SMTPTLSPolicy]
@@ -380,15 +381,15 @@
-ParsedReport = Union[AggregateParsedReport, FailureParsedReport, SMTPTLSParsedReport]
+ParsedReport = AggregateParsedReport | FailureParsedReport | SMTPTLSParsedReport
[docs]
class ParsingResults(TypedDict):
-
aggregate_reports: List[AggregateReport]
-
failure_reports: List[FailureReport]
-
smtp_tls_reports: List[SMTPTLSReport]
+ aggregate_reports: list[AggregateReport]
+ failure_reports: list[FailureReport]
+ smtp_tls_reports: list[SMTPTLSReport]
diff --git a/_modules/parsedmarc/utils.html b/_modules/parsedmarc/utils.html
index 1120492..44801f7 100644
--- a/_modules/parsedmarc/utils.html
+++ b/_modules/parsedmarc/utils.html
@@ -99,7 +99,7 @@
import subprocess
import tempfile
from datetime import datetime, timedelta, timezone
-from typing import Optional, TypedDict, Union, cast
+from typing import TypedDict, cast
import mailparser
from expiringdict import ExpiringDict
@@ -151,8 +151,8 @@
def load_psl_overrides(
*,
always_use_local_file: bool = False,
- local_file_path: Optional[str] = None,
- url: Optional[str] = None,
+ local_file_path: str | None = None,
+ url: str | None = None,
offline: bool = False,
) -> list[str]:
"""
@@ -231,7 +231,7 @@
[docs]
class ReverseDNSService(TypedDict):
name: str
- type: Optional[str]
+ type: str | None
@@ -242,14 +242,14 @@
[docs]
class IPAddressInfo(TypedDict):
ip_address: str
- reverse_dns: Optional[str]
- country: Optional[str]
- base_domain: Optional[str]
- name: Optional[str]
- type: Optional[str]
- asn: Optional[int]
- as_name: Optional[str]
- as_domain: Optional[str]
+ reverse_dns: str | None
+ country: str | None
+ base_domain: str | None
+ name: str | None
+ type: str | None
+ asn: int | None
+ as_name: str | None
+ as_domain: str | None
@@ -276,7 +276,7 @@
[docs]
-
def get_base_domain(domain: str) -> Optional[str]:
+
def get_base_domain(domain: str) -> str | None:
"""
Gets the base domain name for the given domain
@@ -307,8 +307,8 @@
domain: str,
record_type: str,
*,
-
cache: Optional[ExpiringDict] = None,
-
nameservers: Optional[list[str]] = None,
+
cache: ExpiringDict | None = None,
+
nameservers: list[str] | None = None,
timeout: float = DEFAULT_DNS_TIMEOUT,
retries: int = DEFAULT_DNS_MAX_RETRIES,
_attempt: int = 0,
@@ -402,11 +402,11 @@
def get_reverse_dns(
ip_address,
*,
-
cache: Optional[ExpiringDict] = None,
-
nameservers: Optional[list[str]] = None,
+
cache: ExpiringDict | None = None,
+
nameservers: list[str] | None = None,
timeout: float = DEFAULT_DNS_TIMEOUT,
retries: int = DEFAULT_DNS_MAX_RETRIES,
-
) -> Optional[str]:
+
) -> str | None:
"""
Resolves an IP address to a hostname using a reverse DNS query
@@ -514,7 +514,7 @@
-
_IP_DB_PATH: Optional[str] = None
+
_IP_DB_PATH: str | None = None
@@ -522,8 +522,8 @@
def load_ip_db(
*,
always_use_local_file: bool = False,
-
local_file_path: Optional[str] = None,
-
url: Optional[str] = None,
+
local_file_path: str | None = None,
+
url: str | None = None,
offline: bool = False,
) -> None:
"""
@@ -585,10 +585,10 @@
class _IPDatabaseRecord(TypedDict):
-
country: Optional[str]
-
asn: Optional[int]
-
as_name: Optional[str]
-
as_domain: Optional[str]
+
country: str | None
+
asn: int | None
+
as_name: str | None
+
as_domain: str | None
@@ -608,14 +608,14 @@
# here — adding it would be inventing behavior the service doesn't document.
# Authentication uses the documented ``?token=`` query parameter.
_IPINFO_API_URL = "https://api.ipinfo.io/lite"
-
_IPINFO_API_TOKEN: Optional[str] = None
+
_IPINFO_API_TOKEN: str | None = None
_IPINFO_API_TIMEOUT: float = 5.0