Repository: apify/crawlee-python
File: src/crawlee/_types.py (Python)

Code before:
    from crawlee.storages import KeyValueStore
    from crawlee.storages._dataset import ExportToKwargs, GetDataKwargs

    # Workaround for https://github.com/pydantic/pydantic/issues/9445
    J = TypeVar('J', bound='JsonSerializable')
    JsonSerializable: TypeAlias = Union[
        list[J],
        dict[str, J],
        str,
        bool,
        int,
        float,
        None,
    ]
else:
    from pydantic import JsonValue as JsonSerializable

T = TypeVar('T')

HttpMethod: TypeAlias = Literal['GET', 'HEAD', 'POST', 'PUT', 'DELETE', 'CONNECT', 'OPTIONS', 'TRACE', 'PATCH']

HttpPayload: TypeAlias = bytes

RequestTransformAction: TypeAlias = Literal['skip', 'unchanged']

EnqueueStrategy: TypeAlias = Literal['all', 'same_domain', 'same_hostname', 'same_origin']
"""Specify the strategy for determining which links to extract and enqueue.

Strategies:
    all: Enqueue every link encountered, regardless of the target domain. Use this option to ensure that all links,
        including those leading to external websites, are followed.
    same_domain: Enqueue links that share the same domain name as the current page, including any subdomains. This
        strategy is ideal for crawling within the same top-level domain while still allowing for subdomain exploration.
    same_hostname: Enqueue links only if they match the exact hostname of the current page. This is the default
        behavior and restricts the crawl to the current hostname, excluding subdomains.
    same_origin: Enqueue links that share the same origin as the current page. The origin is defined by the
        combination of protocol, domain, and port, ensuring a strict scope for the crawl.
"""


def _normalize_headers(headers: Mapping[str, str]) -> dict[str, str]:
    """Converts all header keys to lowercase, strips whitespace, and returns them sorted by key."""
    normalized_headers = {k.lower().strip(): v.strip() for k, v in headers.items()}
    sorted_headers = sorted(normalized_headers.items())
    return dict(sorted_headers)


@docs_group('Data structures')
class HttpHeaders(RootModel, Mapping[str, str]):
    """A dictionary-like object representing HTTP headers."""


Review comment (suggestion):
This could break if you run the new version with storage populated by the previous version - the `enqueue_strategy` is stored in the request queue and the old version has dashes instead of underscores.

Code after:
    from crawlee.storages import KeyValueStore
    from crawlee.storages._dataset import ExportToKwargs, GetDataKwargs

    # Workaround for https://github.com/pydantic/pydantic/issues/9445
    J = TypeVar('J', bound='JsonSerializable')
    JsonSerializable: TypeAlias = Union[
        list[J],
        dict[str, J],
        str,
        bool,
        int,
        float,
        None,
    ]
else:
    from pydantic import JsonValue as JsonSerializable

T = TypeVar('T')

HttpMethod: TypeAlias = Literal['GET', 'HEAD', 'POST', 'PUT', 'DELETE', 'CONNECT', 'OPTIONS', 'TRACE', 'PATCH']

HttpPayload: TypeAlias = bytes

RequestTransformAction: TypeAlias = Literal['skip', 'unchanged']

EnqueueStrategy: TypeAlias = Literal['all', 'same-domain', 'same-hostname', 'same-origin']
"""Enqueue strategy to be used for determining which links to extract and enqueue."""


def _normalize_headers(headers: Mapping[str, str]) -> dict[str, str]:
    """Converts all header keys to lowercase, strips whitespace, and returns them sorted by key."""
    normalized_headers = {k.lower().strip(): v.strip() for k, v in headers.items()}
    sorted_headers = sorted(normalized_headers.items())
    return dict(sorted_headers)


@docs_group('Data structures')
class HttpHeaders(RootModel, Mapping[str, str]):
    """A dictionary-like object representing HTTP headers."""

    model_config = ConfigDict(populate_by_name=True)

    root: Annotated[
        dict[str, str],
        PlainValidator(lambda value: _normalize_headers(value)),
        Field(default_factory=dict),
    ] = {}  # noqa: RUF012

    def __getitem__(self, key: str) -> str:
        return self.root[key.lower()]