Skip to content

HTTP dereferencing

deref

HTTP dereferencing support for N3 log built-ins.

clear_http_cache 

clear_http_cache() -> None

Clear dereferenced documents, primarily for tests and long-lived apps.

Source code in pyling/deref.py 
42
43
44
def clear_http_cache() -> None:
    """Clear dereferenced documents, primarily for tests and long-lived apps."""
    _HTTP_CACHE.clear()

fetch_http_document 

fetch_http_document(iri: str, *, accept: str = DEFAULT_ACCEPT, timeout: float = DEFAULT_TIMEOUT) -> HttpDocument | None

Fetch an HTTP(S) document using RDFLib-style urllib behavior.

Redirects, Python audit hooks, and custom openers installed with urllib.request.install_opener are honored by urlopen.

Source code in pyling/deref.py 
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
def fetch_http_document(
    iri: str,
    *,
    accept: str = DEFAULT_ACCEPT,
    timeout: float = DEFAULT_TIMEOUT,
) -> HttpDocument | None:
    """Fetch an HTTP(S) document using RDFLib-style urllib behavior.

    Redirects, Python audit hooks, and custom openers installed with
    ``urllib.request.install_opener`` are honored by ``urlopen``.
    """
    document_iri = strip_fragment(iri)
    if urlsplit(document_iri).scheme.lower() not in {"http", "https"}:
        return None
    cache_key = (document_iri, accept)
    if cache_key in _HTTP_CACHE:
        _HTTP_CACHE.move_to_end(cache_key)
        return _HTTP_CACHE[cache_key]

    request = urllib.request.Request(
        document_iri,
        headers={
            "Accept": accept,
            "User-Agent": USER_AGENT,
        },
    )
    try:
        with urllib.request.urlopen(request, timeout=timeout) as response:
            body = response.read(MAX_RESPONSE_BYTES + 1)
            if len(body) > MAX_RESPONSE_BYTES:
                return None
            final_url = response.geturl()
            if urlsplit(final_url).scheme.lower() not in {"http", "https"}:
                return None
            content_type = response.headers.get_content_type()
            charset = response.headers.get_content_charset() or "utf-8"
            result = HttpDocument(body.decode(charset), final_url, content_type)
    except (HTTPError, URLError, LookupError, OSError, UnicodeError, ValueError):
        return None

    _HTTP_CACHE[cache_key] = result
    _HTTP_CACHE.move_to_end(cache_key)
    while len(_HTTP_CACHE) > MAX_CACHE_ENTRIES:
        _HTTP_CACHE.popitem(last=False)
    return result