mirror of
https://github.com/pacnpal/django-anymail.git
synced 2025-12-20 03:41:05 -05:00
823a161927
Replace generic `combine` with specific `merge_dicts_deep`, `merge_dicts_shallow`, `merge_dicts_one_level` or `concat_lists`, depending on appropriate behavior for each message attribute. Fixes merging global `SEND_DEFAULTS` with message `esp_extra` for ESP APIs that use nested payload structures. And clarifies intent for other properties.
681 lines
23 KiB
Python
681 lines
23 KiB
Python
import base64
|
|
import mimetypes
|
|
from base64 import b64encode
|
|
from collections.abc import Mapping, MutableMapping
|
|
from copy import copy, deepcopy
|
|
from email.mime.base import MIMEBase
|
|
from email.utils import formatdate, getaddresses, parsedate_to_datetime, unquote
|
|
from urllib.parse import urlsplit, urlunsplit
|
|
|
|
from django.conf import settings
|
|
from django.core.mail.message import DEFAULT_ATTACHMENT_MIME_TYPE, sanitize_address
|
|
from django.utils.encoding import force_str
|
|
from django.utils.functional import Promise
|
|
from requests.structures import CaseInsensitiveDict
|
|
|
|
from .exceptions import AnymailConfigurationError, AnymailInvalidAddress
|
|
|
|
BASIC_NUMERIC_TYPES = (int, float)
|
|
|
|
|
|
UNSET = type("UNSET", (object,), {}) # Used as non-None default value
|
|
|
|
|
|
def concat_lists(*args):
|
|
"""
|
|
Combines all non-UNSET args, by concatenating lists (or sequence-like types).
|
|
Does not modify any args.
|
|
|
|
>>> concat_lists([1, 2], UNSET, [3, 4], UNSET)
|
|
[1, 2, 3, 4]
|
|
>>> concat_lists([1, 2], None, [3, 4]) # None suppresses earlier args
|
|
[3, 4]
|
|
>>> concat_lists()
|
|
UNSET
|
|
|
|
"""
|
|
result = UNSET
|
|
for value in args:
|
|
if value is None:
|
|
# None is a request to suppress any earlier values
|
|
result = UNSET
|
|
elif value is not UNSET:
|
|
if result is UNSET:
|
|
result = list(value)
|
|
else:
|
|
result = result + list(value) # concatenate sequence-like
|
|
return result
|
|
|
|
|
|
def merge_dicts_shallow(*args):
|
|
"""
|
|
Shallow-merges all non-UNSET args.
|
|
Does not modify any args.
|
|
|
|
>>> merge_dicts_shallow({'a': 1, 'b': 2}, UNSET, {'b': 3, 'c': 4}, UNSET)
|
|
{'a': 1, 'b': 3, 'c': 4}
|
|
>>> merge_dicts_shallow({'a': {'a1': 1, 'a2': 2}}, {'a': {'a1': 11, 'a3': 33}})
|
|
{'a': {'a1': 11, 'a3': 33}}
|
|
>>> merge_dicts_shallow({'a': 1}, None, {'b': 2}) # None suppresses earlier args
|
|
{'b': 2}
|
|
>>> merge_dicts_shallow()
|
|
UNSET
|
|
|
|
"""
|
|
result = UNSET
|
|
for value in args:
|
|
if value is None:
|
|
# None is a request to suppress any earlier values
|
|
result = UNSET
|
|
elif value is not UNSET:
|
|
if result is UNSET:
|
|
result = copy(value)
|
|
else:
|
|
result.update(value)
|
|
return result
|
|
|
|
|
|
def merge_dicts_deep(*args):
|
|
"""
|
|
Deep-merges all non-UNSET args.
|
|
Does not modify any args.
|
|
|
|
>>> merge_dicts_deep({'a': 1, 'b': 2}, UNSET, {'b': 3, 'c': 4}, UNSET)
|
|
{'a': 1, 'b': 3, 'c': 4}
|
|
>>> merge_dicts_deep({'a': {'a1': 1, 'a2': 2}}, {'a': {'a1': 11, 'a3': 33}})
|
|
{'a': {'a1': 11, 'a2': 2, 'a3': 33}}
|
|
>>> merge_dicts_deep({'a': 1}, None, {'b': 2}) # None suppresses earlier args
|
|
{'b': 2}
|
|
>>> merge_dicts_deep()
|
|
UNSET
|
|
|
|
"""
|
|
result = UNSET
|
|
for value in args:
|
|
if value is None:
|
|
# None is a request to suppress any earlier values
|
|
result = UNSET
|
|
elif value is not UNSET:
|
|
if result is UNSET:
|
|
result = deepcopy(value)
|
|
else:
|
|
update_deep(result, value)
|
|
return result
|
|
|
|
|
|
def merge_dicts_one_level(*args):
|
|
"""
|
|
Mixture of merge_dicts_deep and merge_dicts_shallow:
|
|
Deep merges first level, shallow merges second level.
|
|
Does not modify any args.
|
|
|
|
(Useful for {"email": {options...}, ...} style dicts,
|
|
like merge_data: shallow merges the options for each email.)
|
|
"""
|
|
result = UNSET
|
|
for value in args:
|
|
if value is None:
|
|
# None is a request to suppress any earlier values
|
|
result = UNSET
|
|
elif value is not UNSET:
|
|
if result is UNSET:
|
|
result = {}
|
|
for k, v in value.items():
|
|
result.setdefault(k, {}).update(v)
|
|
return result
|
|
|
|
|
|
def last(*args):
|
|
"""Returns the last of its args which is not UNSET.
|
|
|
|
>>> last(1, 2, UNSET, 3, UNSET, UNSET)
|
|
3
|
|
>>> last(1, 2, None, UNSET) # None suppresses earlier args
|
|
UNSET
|
|
>>> last()
|
|
UNSET
|
|
|
|
"""
|
|
for value in reversed(args):
|
|
if value is None:
|
|
# None is a request to suppress any earlier values
|
|
return UNSET
|
|
elif value is not UNSET:
|
|
return value
|
|
return UNSET
|
|
|
|
|
|
def getfirst(dct, keys, default=UNSET):
|
|
"""Returns the value of the first of keys found in dict dct.
|
|
|
|
>>> getfirst({'a': 1, 'b': 2}, ['c', 'a'])
|
|
1
|
|
>>> getfirst({'a': 1, 'b': 2}, ['b', 'a'])
|
|
2
|
|
>>> getfirst({'a': 1, 'b': 2}, ['c'])
|
|
KeyError
|
|
>>> getfirst({'a': 1, 'b': 2}, ['c'], None)
|
|
None
|
|
"""
|
|
for key in keys:
|
|
try:
|
|
return dct[key]
|
|
except KeyError:
|
|
pass
|
|
if default is UNSET:
|
|
raise KeyError("None of %s found in dict" % ", ".join(keys))
|
|
else:
|
|
return default
|
|
|
|
|
|
def update_deep(dct, other):
|
|
"""Merge (recursively) keys and values from dict other into dict dct
|
|
|
|
Works with dict-like objects: dct (and descendants) can be any MutableMapping,
|
|
and other can be any Mapping
|
|
"""
|
|
for key, value in other.items():
|
|
if (
|
|
key in dct
|
|
and isinstance(dct[key], MutableMapping)
|
|
and isinstance(value, Mapping)
|
|
):
|
|
update_deep(dct[key], value)
|
|
else:
|
|
dct[key] = value
|
|
# (like dict.update(), no return value)
|
|
|
|
|
|
def parse_address_list(address_list, field=None):
|
|
"""Returns a list of EmailAddress objects from strings in address_list.
|
|
|
|
Essentially wraps :func:`email.utils.getaddresses` with better error
|
|
messaging and more-useful output objects
|
|
|
|
Note that the returned list might be longer than the address_list param,
|
|
if any individual string contains multiple comma-separated addresses.
|
|
|
|
:param list[str]|str|None|list[None] address_list:
|
|
the address or addresses to parse
|
|
:param str|None field:
|
|
optional description of the source of these addresses, for error message
|
|
:return list[:class:`EmailAddress`]:
|
|
:raises :exc:`AnymailInvalidAddress`:
|
|
"""
|
|
if isinstance(address_list, str) or is_lazy(address_list):
|
|
address_list = [address_list]
|
|
|
|
if address_list is None or address_list == [None]:
|
|
return []
|
|
|
|
# For consistency with Django's SMTP backend behavior, extract all addresses
|
|
# from the list -- which may split comma-seperated strings into multiple addresses.
|
|
# (See django.core.mail.message: EmailMessage.message to/cc/bcc/reply_to handling;
|
|
# also logic for ADDRESS_HEADERS in forbid_multi_line_headers.)
|
|
|
|
# resolve lazy strings:
|
|
address_list_strings = [force_str(address) for address in address_list]
|
|
name_email_pairs = getaddresses(address_list_strings)
|
|
if name_email_pairs == [] and address_list_strings == [""]:
|
|
name_email_pairs = [("", "")] # getaddresses ignores a single empty string
|
|
parsed = [
|
|
EmailAddress(display_name=name, addr_spec=email)
|
|
for (name, email) in name_email_pairs
|
|
]
|
|
|
|
# Sanity-check, and raise useful errors
|
|
for address in parsed:
|
|
if address.username == "" or address.domain == "":
|
|
# Django SMTP allows username-only emails,
|
|
# but they're not meaningful with an ESP
|
|
errmsg = (
|
|
"Invalid email address '{problem}'" " parsed from '{source}'{where}."
|
|
).format(
|
|
problem=address.addr_spec,
|
|
source=", ".join(address_list_strings),
|
|
where=" in `%s`" % field if field else "",
|
|
)
|
|
if len(parsed) > len(address_list):
|
|
errmsg += " (Maybe missing quotes around a display-name?)"
|
|
raise AnymailInvalidAddress(errmsg)
|
|
|
|
return parsed
|
|
|
|
|
|
def parse_single_address(address, field=None):
|
|
"""Parses a single EmailAddress from str address, or raises AnymailInvalidAddress
|
|
|
|
:param str address: the fully-formatted email str to parse
|
|
:param str|None field:
|
|
optional description of the source of this address, for error message
|
|
:return :class:`EmailAddress`: if address contains a single email
|
|
:raises :exc:`AnymailInvalidAddress`: if address contains no or multiple emails
|
|
"""
|
|
parsed = parse_address_list([address], field=field)
|
|
count = len(parsed)
|
|
if count > 1:
|
|
raise AnymailInvalidAddress(
|
|
"Only one email address is allowed;"
|
|
" found {count} in '{address}'{where}.".format(
|
|
count=count, address=address, where=" in `%s`" % field if field else ""
|
|
)
|
|
)
|
|
else:
|
|
return parsed[0]
|
|
|
|
|
|
class EmailAddress:
|
|
"""A sanitized, complete email address with easy access
|
|
to display-name, addr-spec (email), etc.
|
|
|
|
Similar to Python 3.6+ email.headerregistry.Address
|
|
|
|
Instance properties, all read-only:
|
|
:ivar str display_name:
|
|
the address's display-name portion (unqouted, unescaped),
|
|
e.g., 'Display Name, Inc.'
|
|
:ivar str addr_spec:
|
|
the address's addr-spec portion (unquoted, unescaped),
|
|
e.g., 'user@example.com'
|
|
:ivar str username:
|
|
the local part (before the '@') of the addr-spec,
|
|
e.g., 'user'
|
|
:ivar str domain:
|
|
the domain part (after the '@') of the addr-spec,
|
|
e.g., 'example.com'
|
|
|
|
:ivar str address:
|
|
the fully-formatted address, with any necessary quoting and escaping,
|
|
e.g., '"Display Name, Inc." <user@example.com>'
|
|
(also available as `str(EmailAddress)`)
|
|
"""
|
|
|
|
def __init__(self, display_name="", addr_spec=None):
|
|
self._address = None # lazy formatted address
|
|
if addr_spec is None:
|
|
try:
|
|
display_name, addr_spec = display_name # unpack (name,addr) tuple
|
|
except ValueError:
|
|
pass
|
|
|
|
# ESPs should clean or reject addresses containing newlines, but some
|
|
# extra protection can't hurt (and it seems to be a common oversight)
|
|
if "\n" in display_name or "\r" in display_name:
|
|
raise ValueError("EmailAddress display_name cannot contain newlines")
|
|
if "\n" in addr_spec or "\r" in addr_spec:
|
|
raise ValueError("EmailAddress addr_spec cannot contain newlines")
|
|
|
|
self.display_name = display_name
|
|
self.addr_spec = addr_spec
|
|
try:
|
|
self.username, self.domain = addr_spec.split("@", 1)
|
|
# do we need to unquote username?
|
|
except ValueError:
|
|
self.username = addr_spec
|
|
self.domain = ""
|
|
|
|
def __repr__(self):
|
|
return "EmailAddress({display_name!r}, {addr_spec!r})".format(
|
|
display_name=self.display_name, addr_spec=self.addr_spec
|
|
)
|
|
|
|
@property
|
|
def address(self):
|
|
if self._address is None:
|
|
# (you might be tempted to use `encoding=settings.DEFAULT_CHARSET` here,
|
|
# but that always forces the display-name to quoted-printable/base64,
|
|
# even when simple ascii would work fine--and be more readable)
|
|
self._address = self.formataddr()
|
|
return self._address
|
|
|
|
def formataddr(self, encoding=None):
|
|
"""Return a fully-formatted email address, using encoding.
|
|
|
|
This is essentially the same as :func:`email.utils.formataddr`
|
|
on the EmailAddress's name and email properties, but uses
|
|
Django's :func:`~django.core.mail.message.sanitize_address`
|
|
for consistent handling of encoding (a.k.a. charset) and
|
|
proper handling of IDN domain portions.
|
|
|
|
:param str|None encoding:
|
|
the charset to use for the display-name portion;
|
|
default None uses ascii if possible, else 'utf-8'
|
|
(quoted-printable utf-8/base64)
|
|
"""
|
|
return sanitize_address((self.display_name, self.addr_spec), encoding)
|
|
|
|
def __str__(self):
|
|
return self.address
|
|
|
|
|
|
class Attachment:
|
|
"""A normalized EmailMessage.attachments item with additional functionality
|
|
|
|
Normalized to have these properties:
|
|
name: attachment filename; may be None
|
|
content: bytestream
|
|
mimetype: the content type; guessed if not explicit
|
|
inline: bool, True if attachment has a Content-ID header
|
|
content_id: for inline, the Content-ID (*with* <>); may be None
|
|
cid: for inline, the Content-ID *without* <>; may be empty string
|
|
"""
|
|
|
|
def __init__(self, attachment, encoding):
|
|
# Note that an attachment can be either a tuple of (filename, content, mimetype)
|
|
# or a MIMEBase object. (Also, both filename and mimetype may be missing.)
|
|
self._attachment = attachment
|
|
self.encoding = encoding # or check attachment["Content-Encoding"] ???
|
|
self.inline = False
|
|
self.content_id = None
|
|
self.cid = ""
|
|
|
|
if isinstance(attachment, MIMEBase):
|
|
self.name = attachment.get_filename()
|
|
self.content = attachment.get_payload(decode=True)
|
|
if self.content is None:
|
|
self.content = attachment.as_bytes()
|
|
self.mimetype = attachment.get_content_type()
|
|
# Content-Type includes charset if provided
|
|
self.content_type = attachment["Content-Type"]
|
|
|
|
content_disposition = attachment.get_content_disposition()
|
|
if content_disposition == "inline" or (
|
|
not content_disposition and "Content-ID" in attachment
|
|
):
|
|
self.inline = True
|
|
self.content_id = attachment["Content-ID"] # probably including <...>
|
|
if self.content_id is not None:
|
|
self.cid = unquote(self.content_id) # without the <, >
|
|
else:
|
|
(self.name, self.content, self.mimetype) = attachment
|
|
self.content_type = self.mimetype
|
|
|
|
self.name = force_non_lazy(self.name)
|
|
self.content = force_non_lazy(self.content)
|
|
|
|
# Guess missing mimetype from filename, borrowed from
|
|
# django.core.mail.EmailMessage._create_attachment()
|
|
if self.mimetype is None and self.name is not None:
|
|
self.mimetype, _ = mimetypes.guess_type(self.name)
|
|
if self.mimetype is None:
|
|
self.mimetype = DEFAULT_ATTACHMENT_MIME_TYPE
|
|
if self.content_type is None:
|
|
self.content_type = self.mimetype
|
|
|
|
def __repr__(self):
|
|
details = [
|
|
self.mimetype,
|
|
"len={length}".format(length=len(self.content)),
|
|
]
|
|
if self.name:
|
|
details.append("name={name!r}".format(name=self.name))
|
|
if self.inline:
|
|
details.insert(0, "inline")
|
|
details.append(
|
|
"content_id={content_id!r}".format(content_id=self.content_id)
|
|
)
|
|
return "Attachment<{details}>".format(details=", ".join(details))
|
|
|
|
@property
|
|
def b64content(self):
|
|
"""Content encoded as a base64 ascii string"""
|
|
content = self.content
|
|
if isinstance(content, str):
|
|
content = content.encode(self.encoding)
|
|
return b64encode(content).decode("ascii")
|
|
|
|
|
|
def get_anymail_setting(
|
|
name, default=UNSET, esp_name=None, kwargs=None, allow_bare=False
|
|
):
|
|
"""Returns an Anymail option from kwargs or Django settings.
|
|
|
|
Returns first of:
|
|
- kwargs[name] -- e.g., kwargs['api_key'] -- and name key will be popped from kwargs
|
|
- settings.ANYMAIL['<ESP_NAME>_<NAME>'] -- e.g., settings.ANYMAIL['MAILGUN_API_KEY']
|
|
- settings.ANYMAIL_<ESP_NAME>_<NAME> -- e.g., settings.ANYMAIL_MAILGUN_API_KEY
|
|
- settings.<ESP_NAME>_<NAME> (only if allow_bare) -- e.g., settings.MAILGUN_API_KEY
|
|
- default if provided; else raises AnymailConfigurationError
|
|
|
|
If allow_bare, allows settings.<ESP_NAME>_<NAME> without the ANYMAIL_ prefix:
|
|
ANYMAIL = { "MAILGUN_API_KEY": "xyz", ... }
|
|
ANYMAIL_MAILGUN_API_KEY = "xyz"
|
|
MAILGUN_API_KEY = "xyz"
|
|
"""
|
|
|
|
try:
|
|
value = kwargs.pop(name)
|
|
if name in ["username", "password"]:
|
|
# Work around a problem in django.core.mail.send_mail, which calls
|
|
# get_connection(... username=None, password=None) by default.
|
|
# We need to ignore those None defaults (else settings like
|
|
# 'SENDGRID_USERNAME' get unintentionally overridden from kwargs).
|
|
if value is not None:
|
|
return value
|
|
else:
|
|
return value
|
|
except (AttributeError, KeyError):
|
|
pass
|
|
|
|
if esp_name is not None:
|
|
setting = "{}_{}".format(esp_name.upper().replace(" ", "_"), name.upper())
|
|
else:
|
|
setting = name.upper()
|
|
anymail_setting = "ANYMAIL_%s" % setting
|
|
|
|
try:
|
|
return settings.ANYMAIL[setting]
|
|
except (AttributeError, KeyError):
|
|
try:
|
|
return getattr(settings, anymail_setting)
|
|
except AttributeError:
|
|
if allow_bare:
|
|
try:
|
|
return getattr(settings, setting)
|
|
except AttributeError:
|
|
pass
|
|
if default is UNSET:
|
|
message = "You must set %s or ANYMAIL = {'%s': ...}" % (
|
|
anymail_setting,
|
|
setting,
|
|
)
|
|
if allow_bare:
|
|
message += " or %s" % setting
|
|
message += " in your Django settings"
|
|
raise AnymailConfigurationError(message) from None
|
|
else:
|
|
return default
|
|
|
|
|
|
def collect_all_methods(cls, method_name):
|
|
"""Return list of all `method_name` methods for cls and its superclass chain.
|
|
|
|
List is in MRO order, with no duplicates. Methods are unbound.
|
|
|
|
(This is used to simplify mixins and subclasses that contribute to a method set,
|
|
without requiring superclass chaining, and without requiring cooperating
|
|
superclasses.)
|
|
"""
|
|
methods = []
|
|
for ancestor in cls.__mro__:
|
|
try:
|
|
validator = getattr(ancestor, method_name)
|
|
except AttributeError:
|
|
pass
|
|
else:
|
|
if validator not in methods:
|
|
methods.append(validator)
|
|
return methods
|
|
|
|
|
|
def querydict_getfirst(qdict, field, default=UNSET):
|
|
"""
|
|
Like :func:`django.http.QueryDict.get`,
|
|
but returns *first* value of multi-valued field.
|
|
|
|
>>> from django.http import QueryDict
|
|
>>> q = QueryDict('a=1&a=2&a=3')
|
|
>>> querydict_getfirst(q, 'a')
|
|
'1'
|
|
>>> q.get('a')
|
|
'3'
|
|
>>> q['a']
|
|
'3'
|
|
|
|
You can bind this to a QueryDict instance using the "descriptor protocol":
|
|
>>> q.getfirst = querydict_getfirst.__get__(q)
|
|
>>> q.getfirst('a')
|
|
'1'
|
|
"""
|
|
# (Why not instead define a QueryDict subclass with this method? Because there's
|
|
# no simple way to efficiently initialize a QueryDict subclass with the contents
|
|
# of an existing instance.)
|
|
values = qdict.getlist(field)
|
|
if len(values) > 0:
|
|
return values[0]
|
|
elif default is not UNSET:
|
|
return default
|
|
else:
|
|
return qdict[field] # raise appropriate KeyError
|
|
|
|
|
|
def rfc2822date(dt):
|
|
"""Turn a datetime into a date string as specified in RFC 2822."""
|
|
# This is almost the equivalent of Python's email.utils.format_datetime,
|
|
# but treats naive datetimes as local rather than "UTC with no information ..."
|
|
timeval = dt.timestamp()
|
|
return formatdate(timeval, usegmt=True)
|
|
|
|
|
|
def angle_wrap(s):
|
|
"""Return s surrounded by angle brackets, added only if necessary"""
|
|
# This is the inverse behavior of email.utils.unquote
|
|
# (which you might think email.utils.quote would do, but it doesn't)
|
|
if len(s) > 0:
|
|
if s[0] != "<":
|
|
s = "<" + s
|
|
if s[-1] != ">":
|
|
s = s + ">"
|
|
return s
|
|
|
|
|
|
def is_lazy(obj):
|
|
"""Return True if obj is a Django lazy object."""
|
|
# See django.utils.functional.lazy. (This appears to be preferred
|
|
# to checking for `not isinstance(obj, str)`.)
|
|
return isinstance(obj, Promise)
|
|
|
|
|
|
def force_non_lazy(obj):
|
|
"""
|
|
If obj is a Django lazy object, return it coerced to text;
|
|
otherwise return it unchanged.
|
|
|
|
(Similar to django.utils.encoding.force_text, but doesn't alter non-text objects.)
|
|
"""
|
|
if is_lazy(obj):
|
|
return str(obj)
|
|
|
|
return obj
|
|
|
|
|
|
def force_non_lazy_list(obj):
|
|
"""Return a (shallow) copy of sequence obj, with all values forced non-lazy."""
|
|
try:
|
|
return [force_non_lazy(item) for item in obj]
|
|
except (AttributeError, TypeError):
|
|
return force_non_lazy(obj)
|
|
|
|
|
|
def force_non_lazy_dict(obj):
|
|
"""Return a (deep) copy of dict obj, with all values forced non-lazy."""
|
|
try:
|
|
return {key: force_non_lazy_dict(value) for key, value in obj.items()}
|
|
except (AttributeError, TypeError):
|
|
return force_non_lazy(obj)
|
|
|
|
|
|
def get_request_basic_auth(request):
|
|
"""Returns HTTP basic auth string sent with request, or None.
|
|
|
|
If request includes basic auth, result is string 'username:password'.
|
|
"""
|
|
try:
|
|
authtype, authdata = request.META["HTTP_AUTHORIZATION"].split()
|
|
if authtype.lower() == "basic":
|
|
return base64.b64decode(authdata).decode("utf-8")
|
|
except (IndexError, KeyError, TypeError, ValueError):
|
|
pass
|
|
return None
|
|
|
|
|
|
def get_request_uri(request):
|
|
"""Returns the "exact" url used to call request.
|
|
|
|
Like :func:`django.http.request.HTTPRequest.build_absolute_uri`,
|
|
but also inlines HTTP basic auth, if present.
|
|
"""
|
|
url = request.build_absolute_uri()
|
|
basic_auth = get_request_basic_auth(request)
|
|
if basic_auth is not None:
|
|
# must reassemble url with auth
|
|
parts = urlsplit(url)
|
|
url = urlunsplit(
|
|
(
|
|
parts.scheme,
|
|
basic_auth + "@" + parts.netloc,
|
|
parts.path,
|
|
parts.query,
|
|
parts.fragment,
|
|
)
|
|
)
|
|
return url
|
|
|
|
|
|
def parse_rfc2822date(s):
|
|
"""Parses an RFC-2822 formatted date string into a datetime.datetime
|
|
|
|
Returns None if string isn't parseable. Returned datetime will be naive
|
|
if string doesn't include known timezone offset; aware if it does.
|
|
|
|
(Same as Python 3 email.utils.parsedate_to_datetime, with improved
|
|
handling for unparseable date strings.)
|
|
"""
|
|
try:
|
|
return parsedate_to_datetime(s)
|
|
except (IndexError, TypeError, ValueError):
|
|
# despite the docs, parsedate_to_datetime often dies on unparseable input
|
|
return None
|
|
|
|
|
|
class CaseInsensitiveCasePreservingDict(CaseInsensitiveDict):
|
|
"""A dict with case-insensitive keys, which preserves the *first* key set.
|
|
|
|
>>> cicpd = CaseInsensitiveCasePreservingDict()
|
|
>>> cicpd["Accept"] = "application/text+xml"
|
|
>>> cicpd["accEPT"] = "application/json"
|
|
>>> cicpd["accept"]
|
|
"application/json"
|
|
>>> cicpd.keys()
|
|
["Accept"]
|
|
|
|
Compare to CaseInsensitiveDict, which preserves *last* key set:
|
|
>>> cid = CaseInsensitiveCasePreservingDict()
|
|
>>> cid["Accept"] = "application/text+xml"
|
|
>>> cid["accEPT"] = "application/json"
|
|
>>> cid.keys()
|
|
["accEPT"]
|
|
"""
|
|
|
|
def __setitem__(self, key, value):
|
|
_k = key.lower()
|
|
try:
|
|
# retrieve earlier matching key, if any
|
|
key, _ = self._store[_k]
|
|
except KeyError:
|
|
pass
|
|
self._store[_k] = (key, value)
|
|
|
|
def copy(self):
|
|
return self.__class__(self._store.values())
|