11292583069

Committed 11 Oct 2024 12:44PM UTC coverage: 96.465%. First build

Build # 11292583069

Build Type

Pull #221

github

Committed by

Wim-De-Clercq

Commit Message

Backend upgrade.

Issue #220

Pull Request Pull Request #221: Feature/220 backend

Run Details

203 of 212 new or added lines in 7 files covered. (95.75%)

655 of 679 relevant lines covered (96.47%)

2.89 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

95.74

/crabpy_pyramid/utils.py

# -*- coding: utf-8 -*-
"""
Utility functions to help with range handling.

.. versionadded:: 0.1.0
"""

import re
from typing import Literal
from typing import Sequence


MAX_NUMBER_ITEMS = 5000


def parse_range_header(range) -> dict[str, int] | Literal[False]:
    """
    Parse a range header as used by the dojo Json Rest store.

    :param str range: The content of the range header to be parsed.
        eg. `items=0-9`
    :returns: A dict with keys start, finish and number or `False` if the
        range is invalid.
    """
    match = re.match("^items=([0-9]+)-([0-9]+)$", range)

    if match:
        start = int(match.group(1))
        finish = int(match.group(2))

        if finish < start:
            finish = start
        return {"start": start, "finish": finish, "count": finish - start + 1}
    else:
        return False


def range_return(request, items: Sequence):
    """
    Determine what range of objects to return.

    Will check fot both `Range` and `X-Range` headers in the request and
    set both `Content-Range` and 'X-Content-Range' headers.

    :rtype: list
    """
    if "Range" in request.headers:
        range = parse_range_header(request.headers["Range"])
        if not range:
            raise ValueError(request.headers["Range"])
    elif "X-Range" in request.headers:
        range = parse_range_header(request.headers["X-Range"])
        if not range:
            raise ValueError(request.headers["X-Range"])
    else:
        range = {"start": 0, "finish": MAX_NUMBER_ITEMS - 1, "count": MAX_NUMBER_ITEMS}
    filtered = items[range["start"] : range["finish"] + 1]
    if len(filtered) < range["count"]:
        # Something was stripped, deal with it
        range["count"] = len(filtered)
        range["finish"] = range["start"] + range["count"] - 1
    if range["finish"] - range["start"] + 1 >= MAX_NUMBER_ITEMS:
        range["finish"] = range["start"] + MAX_NUMBER_ITEMS - 1
        filtered = items[range["start"] : range["finish"] + 1]

    request.response.headers["Content-Range"] = "items %d-%d/%d" % (
        range["start"],
        range["finish"],
        len(items),
    )
    request.response.headers["X-Content-Range"] = request.response.headers[
        "Content-Range"
    ]
    return filtered


def set_http_caching(request, gateway="crab", region="permanent"):
    """
    Set an HTTP Cache Control header on a request.

    :param pyramid.request.Request request: Request to set headers on.
    :param str gateway: What gateway are we caching for? Defaults to `crab`.
    :param str region: What caching region to use? Defaults to `permanent`.
    :rtype: pyramid.request.Request
    """
    crabpy_exp = request.registry.settings.get(
        "crabpy.%s.cache_config.%s.expiration_time" % (gateway, region), None
    )
    if crabpy_exp is None:
        return request
    ctime = int(int(crabpy_exp) * 1.05)
    request.response.cache_expires(ctime, public=True)
    return request


def filter_settings(settings, prefix):
    """
    Filter all settings to only return settings that start with a certain
    prefix.

    :param dict settings: A settings dictionary.
    :param str prefix: A prefix.
    """
    ret = {}
    for skey in settings.keys():
        if skey.startswith(prefix):
            key = skey[len(prefix) :]
            ret[key] = settings[skey]
    return ret

1	# -- coding: utf-8 --
2	"""
3	Utility functions to help with range handling.
4
5	.. versionadded:: 0.1.0
6	"""
7
8	import re	3✔
9	from typing import Literal	3✔
10	from typing import Sequence	3✔
11
12
13	MAX_NUMBER_ITEMS = 5000	3✔
14
15
16	def parse_range_header(range) -> dict[str, int] \| Literal[False]:	3✔
17	"""
18	Parse a range header as used by the dojo Json Rest store.
19
20	:param str range: The content of the range header to be parsed.
21	eg. `items=0-9`
22	:returns: A dict with keys start, finish and number or `False` if the
23	range is invalid.
24	"""
25	match = re.match("^items=([0-9]+)-([0-9]+)$", range)	3✔
26
27	if match:	3✔
28	start = int(match.group(1))	3✔
29	finish = int(match.group(2))	3✔
30
31	if finish < start:	3✔
32	finish = start	3✔
33	return {"start": start, "finish": finish, "count": finish - start + 1}	3✔
34	else:
35	return False	3✔
36
37
38	def range_return(request, items: Sequence):	3✔
39	"""
40	Determine what range of objects to return.
41
42	Will check fot both `Range` and `X-Range` headers in the request and
43	set both `Content-Range` and 'X-Content-Range' headers.
44
45	:rtype: list
46	"""
47	if "Range" in request.headers:	3✔
48	range = parse_range_header(request.headers["Range"])	3✔
49	if not range:	3✔
NEW 50	raise ValueError(request.headers["Range"])	×
51	elif "X-Range" in request.headers:	3✔
52	range = parse_range_header(request.headers["X-Range"])	3✔
53	if not range:	3✔
NEW 54	raise ValueError(request.headers["X-Range"])	×
55	else:
56	range = {"start": 0, "finish": MAX_NUMBER_ITEMS - 1, "count": MAX_NUMBER_ITEMS}	3✔
57	filtered = items[range["start"] : range["finish"] + 1]	3✔
58	if len(filtered) < range["count"]:	3✔
59	# Something was stripped, deal with it
60	range["count"] = len(filtered)	3✔
61	range["finish"] = range["start"] + range["count"] - 1	3✔
62	if range["finish"] - range["start"] + 1 >= MAX_NUMBER_ITEMS:	3✔
63	range["finish"] = range["start"] + MAX_NUMBER_ITEMS - 1	3✔
64	filtered = items[range["start"] : range["finish"] + 1]	3✔
65
66	request.response.headers["Content-Range"] = "items %d-%d/%d" % (	3✔
67	range["start"],
68	range["finish"],
69	len(items),
70	)
71	request.response.headers["X-Content-Range"] = request.response.headers[	3✔
72	"Content-Range"
73	]
74	return filtered	3✔
75
76
77	def set_http_caching(request, gateway="crab", region="permanent"):	3✔
78	"""
79	Set an HTTP Cache Control header on a request.
80
81	:param pyramid.request.Request request: Request to set headers on.
82	:param str gateway: What gateway are we caching for? Defaults to `crab`.
83	:param str region: What caching region to use? Defaults to `permanent`.
84	:rtype: pyramid.request.Request
85	"""
86	crabpy_exp = request.registry.settings.get(	3✔
87	"crabpy.%s.cache_config.%s.expiration_time" % (gateway, region), None
88	)
89	if crabpy_exp is None:	3✔
90	return request	3✔
91	ctime = int(int(crabpy_exp) * 1.05)	3✔
92	request.response.cache_expires(ctime, public=True)	3✔
93	return request	3✔
94
95
96	def filter_settings(settings, prefix):	3✔
97	"""
98	Filter all settings to only return settings that start with a certain
99	prefix.
100
101	:param dict settings: A settings dictionary.
102	:param str prefix: A prefix.
103	"""
104	ret = {}	3✔
105	for skey in settings.keys():	3✔
106	if skey.startswith(prefix):	3✔
107	key = skey[len(prefix) :]	3✔
108	ret[key] = settings[skey]	3✔
109	return ret	3✔

OnroerendErfgoed / crabpy_pyramid / 11292583069

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous