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

98.39

/crabpy_pyramid/capakey.py

"""
The capakey component for crabpy pyramid.

Known settings:
crabpy.capakey.cache_config.*
"""

import functools
import logging
from dataclasses import dataclass
from typing import Any
from typing import Mapping

from crabpy.gateway.capakey import CapakeyRestGateway
from pyramid.config import Configurator
from pyramid.registry import Registry
from pyramid.request import Request
from zope.interface import Interface

from crabpy_pyramid.renderers.capakey import json_item_renderer
from crabpy_pyramid.renderers.capakey import json_list_renderer


LOG = logging.getLogger(__name__)


class ICapakey(Interface):
    pass


@dataclass
class ParsedSettings:
    cache_config: dict[str, str] | None


def includeme(config: Configurator):
    LOG.info("Adding capakey to the application.")
    add_renderers(config)
    build_capakey(config)
    add_routes(config)
    add_views(config)


def parse_settings(settings: Mapping[str, Any]) -> ParsedSettings:
    # cache configuration
    prefix = "crabpy.capakey.cache_config."
    cutoff = len(prefix)
    cache_config = {}
    for key, value in settings.items():
        if key.startswith(prefix):
            cache_config[key[cutoff:]] = value

    LOG.debug(f"{ParsedSettings=}")
    return ParsedSettings(cache_config=cache_config or None)


def add_views(config: Configurator):
    config.scan("crabpy_pyramid.views.capakey")


def add_routes(config: Configurator):
    config.include("crabpy_pyramid.routes.capakey")


def add_renderers(config: Configurator):
    config.add_renderer("capakey_listjson", json_list_renderer)
    config.add_renderer("capakey_itemjson", json_item_renderer)


@functools.singledispatch
def get_capakey(arg) -> CapakeyRestGateway:
    raise NotImplementedError(f"Invalid argument {arg}. Pass a request or registry.")


@get_capakey.register
def _(registry: Registry) -> CapakeyRestGateway:
    return registry.queryUtility(ICapakey)


@get_capakey.register
def _(request: Request) -> CapakeyRestGateway:
    return request.registry.queryUtility(ICapakey)


def build_capakey(config: Configurator) -> CapakeyRestGateway:
    """
    Create an capakey Gateway and set it up for pyramid usage.

    This method does 3 things:
    - Create the gateway
    - Register it as a utility in the registry under `ICapakey`
    - Add a request method `capakey_gateway`
    """
    registry: Registry = config.registry  # type: ignore
    capakey = get_capakey(registry)
    if capakey is not None:
        return capakey

    # Start building
    parsed_settings = parse_settings(registry.settings)
    kwargs = {}
    if parsed_settings.cache_config:
        kwargs["cache_config"] = parsed_settings.cache_config

    gateway = CapakeyRestGateway(**kwargs)

    registry.registerUtility(gateway, ICapakey)
    config.add_request_method(get_capakey, "capakey_gateway")
    return gateway

1	"""
2	The capakey component for crabpy pyramid.
3
4	Known settings:
5	crabpy.capakey.cache_config.*
6	"""
7
8	import functools	3✔
9	import logging	3✔
10	from dataclasses import dataclass	3✔
11	from typing import Any	3✔
12	from typing import Mapping	3✔
13
14	from crabpy.gateway.capakey import CapakeyRestGateway	3✔
15	from pyramid.config import Configurator	3✔
16	from pyramid.registry import Registry	3✔
17	from pyramid.request import Request	3✔
18	from zope.interface import Interface	3✔
19
20	from crabpy_pyramid.renderers.capakey import json_item_renderer	3✔
21	from crabpy_pyramid.renderers.capakey import json_list_renderer	3✔
22
23
24	LOG = logging.getLogger(__name__)	3✔
25
26
27	class ICapakey(Interface):	3✔
28	pass	3✔
29
30
31	@dataclass	3✔
32	class ParsedSettings:	3✔
33	cache_config: dict[str, str] \| None	3✔
34
35
36	def includeme(config: Configurator):	3✔
37	LOG.info("Adding capakey to the application.")	3✔
38	add_renderers(config)	3✔
39	build_capakey(config)	3✔
40	add_routes(config)	3✔
41	add_views(config)	3✔
42
43
44	def parse_settings(settings: Mapping[str, Any]) -> ParsedSettings:	3✔
45	# cache configuration
46	prefix = "crabpy.capakey.cache_config."	3✔
47	cutoff = len(prefix)	3✔
48	cache_config = {}	3✔
49	for key, value in settings.items():	3✔
50	if key.startswith(prefix):	3✔
51	cache_config[key[cutoff:]] = value	3✔
52
53	LOG.debug(f"{ParsedSettings=}")	3✔
54	return ParsedSettings(cache_config=cache_config or None)	3✔
55
56
57	def add_views(config: Configurator):	3✔
58	config.scan("crabpy_pyramid.views.capakey")	3✔
59
60
61	def add_routes(config: Configurator):	3✔
62	config.include("crabpy_pyramid.routes.capakey")	3✔
63
64
65	def add_renderers(config: Configurator):	3✔
66	config.add_renderer("capakey_listjson", json_list_renderer)	3✔
67	config.add_renderer("capakey_itemjson", json_item_renderer)	3✔
68
69
70	@functools.singledispatch	3✔
71	def get_capakey(arg) -> CapakeyRestGateway:	3✔
NEW 72	raise NotImplementedError(f"Invalid argument {arg}. Pass a request or registry.")	×
73
74
75	@get_capakey.register	3✔
76	def _(registry: Registry) -> CapakeyRestGateway:	3✔
77	return registry.queryUtility(ICapakey)	3✔
78
79
80	@get_capakey.register	3✔
81	def _(request: Request) -> CapakeyRestGateway:	3✔
82	return request.registry.queryUtility(ICapakey)	3✔
83
84
85	def build_capakey(config: Configurator) -> CapakeyRestGateway:	3✔
86	"""
87	Create an capakey Gateway and set it up for pyramid usage.
88
89	This method does 3 things:
90	- Create the gateway
91	- Register it as a utility in the registry under `ICapakey`
92	- Add a request method `capakey_gateway`
93	"""
94	registry: Registry = config.registry # type: ignore	3✔
95	capakey = get_capakey(registry)	3✔
96	if capakey is not None:	3✔
97	return capakey	3✔
98
99	# Start building
100	parsed_settings = parse_settings(registry.settings)	3✔
101	kwargs = {}	3✔
102	if parsed_settings.cache_config:	3✔
103	kwargs["cache_config"] = parsed_settings.cache_config	3✔
104
105	gateway = CapakeyRestGateway(**kwargs)	3✔
106
107	registry.registerUtility(gateway, ICapakey)	3✔
108	config.add_request_method(get_capakey, "capakey_gateway")	3✔
109	return gateway	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