lagom.experimental.integrations.django._update_container_for_request() - Code Metrics - meadsteve/lagom - Measure and Improve Code Quality continuously with Scrutinizer

_update_container_for_request() A
last analyzed 2024-11-22 07:23 UTC

↳ Parent: lagom.experimental.integrations.django

Complexity

Conditions

Size

Total Lines	9
Code Lines	6

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	1
CRAP Score	7.608

Importance

Changes

Metric	Value
cc	3
eloc	6
nop	3
dl	0
loc	9
ccs	1
cts	5
cp	0.2
crap	7.608
rs	10
c	0
b	0
f	0

"""
Django integration.

Usage:


```
#views.py

from .dependency_config import dependencies


@dependencies.bind_view
def index(request, dep: SomeDep=injectable, questions: DjangoModel[Question]=injectable):
    new_question = questions.new(question_text="What's next?", pub_date=timezone.now())
    new_question.save()
    return HttpResponse(f"plain old function: {dep.message} with {questions.objects.all().count()} questions")


@dependencies.bind_view
class CBVexample(View):
    def get(self, request, dep: SomeDep=injectable, settings: DjangoSettings):
        # do something with settings
        return HttpResponse(f"Class based: {dep.message}")

```

```
# dependency_config.py
# a per app dep injection container

# Lists all models that should be available
container = Container()
container[SomeService] = SomeService("connection details etc")

dependencies = DjangoIntegration(container, models=[Question], request_singletons=[SomeCache])
```

"""

import types
from typing import TypeVar, Generic, List, Type, Optional, Any

from django.db.models import Manager, Model
from django.http import HttpRequest
from django.views import View

from lagom.interfaces import ExtendableContainer, WriteableContainer
from lagom.definitions import ConstructionWithoutContainer, PlainInstance

M = TypeVar("M", bound=Model)


class _Managers(Generic[M]):
    """
    Wraps around a django model and provides access to the class properties.
    The intention is that all the Manager objects can be accessed via this.
    """

    def __init__(self, model: Type[M]):
        self.model = model

    def __getattr__(self, item) -> Manager:
        if not hasattr(self.model, item):
            raise KeyError(
                f"Django model {self.model.__name__} does not define a property {item}"
            )
        return getattr(self.model, item)


class _DjangoSettings:
    pass


DjangoSettings: Any = _DjangoSettings


class DjangoModel(Generic[M]):
    """Wrapper around a django model for injection hinting
    Usage:
        container = DjangoContainer(models=[Question])

        @bind_to_container(container)
        def load_first_question(questions: DjangoModel[Question]):
            return questions.objects.first()

    """

    model: Type[M]
    managers: _Managers[M]

    def __init__(self, model: Type[M]):
        """

        :param model: The django model class
        """
        self.model = model
        self.managers = _Managers(self.model)

    @property
    def objects(self) -> Manager:
        """Equivalent to MyModel.objects

        :return:
        """
        return self.managers.objects

    def new(self, **kwargs) -> M:
        """Equivalent to MyModel(**kwargs)

        :param kwargs:
        :return:
        """
        return self.model(**kwargs)


class DjangoIntegration:
    """
    Same behaviour as the basic container bug provides a view method which
    should be used as a decorator to wrap views. Once wrapped
    the view can reference dependencies which will be auto-wired.
    """

    _request_singletons: List[Type]
    _container: WriteableContainer

    def __init__(
        self,
        container: ExtendableContainer,
        models: Optional[List[Type[Model]]] = None,
        request_singletons: Optional[List[Type]] = None,
    ):
        """
        :param models: List of models which should be available for injection
        :param request_singletons:
        :param container:
        """
        self._container = container.clone()
        self._request_singletons = request_singletons or []
        self._container.define(
            DjangoSettings, ConstructionWithoutContainer(self._load_settings)
        )
        for model in models or []:
            self._container.define(DjangoModel[model], DjangoModel(model))  # type: ignore

    def bind_view(self, view):
        """
        Takes either a plain function view or a class based view
        binds it to the container then returns something that can
        be used in a django url definition. Only arguments with
        a default of "lagom.injectable" will be bound to the container.
        :param view:
        :return:
        """
        if isinstance(view, types.FunctionType):
            # Plain old function can be bound to the container
            return self._container.partial(
                view,
                shared=self._request_singletons,
                container_updater=_update_container_for_request,
            )
        return self._bind_view_methods_to_container(view)

    def magic_bind_view(self, view):
        """
        Takes either a plain function view or a class based view
        binds it to the container then returns something that can
        be used in a django url definition
        :param view:
        :return:
        """
        if isinstance(view, types.FunctionType):
            # Plain old function can be bound to the container
            return self._container.magic_partial(
                view,
                shared=self._request_singletons,
                skip_pos_up_to=1,
                container_updater=_update_container_for_request,
            )
        return self._bind_view_methods_to_container(view, magic=True)

    def _bind_view_methods_to_container(self, view, magic=False):
        for method in View.http_method_names:
            if hasattr(view, method):
                if magic:
                    bound_func = self._container.magic_partial(
                        getattr(view, method),
                        shared=self._request_singletons,
                        skip_pos_up_to=1,
                        container_updater=_update_container_for_request,
                    )
                else:
                    bound_func = self._container.partial(
                        getattr(view, method),
                        shared=self._request_singletons,
                        container_updater=_update_container_for_request,
                    )
                setattr(
                    view,
                    method,
                    bound_func,
                )
        return view

    @staticmethod
    def _load_settings():
        from django.conf import settings

        return settings


def _update_container_for_request(
    container: WriteableContainer, call_args, call_kwargs
):
    # The first arg is probably a request.
    # lets make that available for injection
    if len(call_args) > 0:
        request = call_args[0]
        if isinstance(request, HttpRequest):

            container[HttpRequest] = PlainInstance(request)


1		"""
2		Django integration.
3
4		Usage:
5
6
7		```
8		#views.py
9
10		from .dependency_config import dependencies
11
12
13		@dependencies.bind_view
14		def index(request, dep: SomeDep=injectable, questions: DjangoModel[Question]=injectable):
15		new_question = questions.new(question_text="What's next?", pub_date=timezone.now())
16		new_question.save()
17		return HttpResponse(f"plain old function: {dep.message} with {questions.objects.all().count()} questions")
18
19
20		@dependencies.bind_view
21		class CBVexample(View):
22		def get(self, request, dep: SomeDep=injectable, settings: DjangoSettings):
23		# do something with settings
24		return HttpResponse(f"Class based: {dep.message}")
25
26		```
27
28		```
29		# dependency_config.py
30		# a per app dep injection container
31
32		# Lists all models that should be available
33		container = Container()
34		container[SomeService] = SomeService("connection details etc")
35
36		dependencies = DjangoIntegration(container, models=[Question], request_singletons=[SomeCache])
37		```
38
39		"""
40
41	1	import types
42	1	from typing import TypeVar, Generic, List, Type, Optional, Any
43
44	1	from django.db.models import Manager, Model
45	1	from django.http import HttpRequest
46	1	from django.views import View
47
48	1	from lagom.interfaces import ExtendableContainer, WriteableContainer
49	1	from lagom.definitions import ConstructionWithoutContainer, PlainInstance
50
51	1	M = TypeVar("M", bound=Model)
52
53
54	1	class _Managers(Generic[M]):
55		"""
56		Wraps around a django model and provides access to the class properties.
57		The intention is that all the Manager objects can be accessed via this.
58		"""
59
60	1	def __init__(self, model: Type[M]):
61	1	self.model = model
62
63	1	def __getattr__(self, item) -> Manager:
64	1	if not hasattr(self.model, item):
65	1	raise KeyError(
66		f"Django model {self.model.__name__} does not define a property {item}"
67		)
68	1	return getattr(self.model, item)
69
70
71	1	class _DjangoSettings:
72	1	pass
73
74
75	1	DjangoSettings: Any = _DjangoSettings
76
77
78	1	class DjangoModel(Generic[M]):
79		"""Wrapper around a django model for injection hinting
80		Usage:
81		container = DjangoContainer(models=[Question])
82
83		@bind_to_container(container)
84		def load_first_question(questions: DjangoModel[Question]):
85		return questions.objects.first()
86
87		"""
88
89	1	model: Type[M]
90	1	managers: _Managers[M]
91
92	1	def __init__(self, model: Type[M]):
93		"""
94
95		:param model: The django model class
96		"""
97	1	self.model = model
98	1	self.managers = _Managers(self.model)
99
100	1	@property
101	1	def objects(self) -> Manager:
102		"""Equivalent to MyModel.objects
103
104		:return:
105		"""
106	1	return self.managers.objects
107
108	1	def new(self, **kwargs) -> M:
109		"""Equivalent to MyModel(**kwargs)
110
111		:param kwargs:
112		:return:
113		"""
114	1	return self.model(**kwargs)
115
116
117	1	class DjangoIntegration:
118		"""
119		Same behaviour as the basic container bug provides a view method which
120		should be used as a decorator to wrap views. Once wrapped
121		the view can reference dependencies which will be auto-wired.
122		"""
123
124	1	_request_singletons: List[Type]
125	1	_container: WriteableContainer
126
127	1	def __init__(
128		self,
129		container: ExtendableContainer,
130		models: Optional[List[Type[Model]]] = None,
131		request_singletons: Optional[List[Type]] = None,
132		):
133		"""
134		:param models: List of models which should be available for injection
135		:param request_singletons:
136		:param container:
137		"""
138	1	self._container = container.clone()
139	1	self._request_singletons = request_singletons or []
140	1	self._container.define(
141		DjangoSettings, ConstructionWithoutContainer(self._load_settings)
142		)
143	1	for model in models or []:
144	1	self._container.define(DjangoModel[model], DjangoModel(model)) # type: ignore
145
146	1	def bind_view(self, view):
147		"""
148		Takes either a plain function view or a class based view
149		binds it to the container then returns something that can
150		be used in a django url definition. Only arguments with
151		a default of "lagom.injectable" will be bound to the container.
152		:param view:
153		:return:
154		"""
155		if isinstance(view, types.FunctionType):
156		# Plain old function can be bound to the container
157		return self._container.partial(
158		view,
159		shared=self._request_singletons,
160		container_updater=_update_container_for_request,
161		)
162		return self._bind_view_methods_to_container(view)
163
164	1	def magic_bind_view(self, view):
165		"""
166		Takes either a plain function view or a class based view
167		binds it to the container then returns something that can
168		be used in a django url definition
169		:param view:
170		:return:
171		"""
172		if isinstance(view, types.FunctionType):
173		# Plain old function can be bound to the container
174		return self._container.magic_partial(
175		view,
176		shared=self._request_singletons,
177		skip_pos_up_to=1,
178		container_updater=_update_container_for_request,
179		)
180		return self._bind_view_methods_to_container(view, magic=True)
181
182	1	def _bind_view_methods_to_container(self, view, magic=False):
183		for method in View.http_method_names:
184		if hasattr(view, method):
185		if magic:
186		bound_func = self._container.magic_partial(
187		getattr(view, method),
188		shared=self._request_singletons,
189		skip_pos_up_to=1,
190		container_updater=_update_container_for_request,
191		)
192		else:
193		bound_func = self._container.partial(
194		getattr(view, method),
195		shared=self._request_singletons,
196		container_updater=_update_container_for_request,
197		)
198		setattr(
199		view,
200		method,
201		bound_func,
202		)
203		return view
204
205	1	@staticmethod
206	1	def _load_settings():
207		from django.conf import settings
208
209		return settings
210
211
212	1	def _update_container_for_request(
213		container: WriteableContainer, call_args, call_kwargs
214		):
215		# The first arg is probably a request.
216		# lets make that available for injection
217		if len(call_args) > 0:
218		request = call_args[0]
219		if isinstance(request, HttpRequest):
		0 ignored issues – show Comprehensibility Best Practice introduced 2020-11-29 20:53 UTC by Report Bug Copy Issue Report The variable `HttpRequest` does not seem to be defined. Loading history...
220		container[HttpRequest] = PlainInstance(request)
221

meadsteve / lagom

_update_container_for_request() A last analyzed 2024-11-22 07:23 UTC

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like

_update_container_for_request() A
last analyzed 2024-11-22 07:23 UTC