13814784687

Committed 12 Mar 2025 03:08PM UTC coverage: 97.609% (+0.001%) from 97.608%

Build # 13814784687

Build Type

push

github

Committed by

idlesign

Commit Message

 Runtime. Allow Mule and Farm objects to be passed as target for time and cron functions.

Run Details

23 of 24 new or added lines in 2 files covered. (95.83%)

2 existing lines in 1 file now uncovered.

4613 of 4726 relevant lines covered (97.61%)

2.89 hits per line

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

96.97

/uwsgiconf/options/python.py

import sys

from ..base import OptionsGroup
from ..exceptions import ConfigurationError
from ..typehints import Strint, Strlist, Strpath

AUTO = -1


class Python(OptionsGroup):
    """Python plugin options.

    .. note:: By default the plugin does not initialize the GIL.
        This means your app-generated threads will not run.
        If you need threads, remember to enable them with ``enable_threads``.

    """
    plugin = True

    def set_basic_params(
            self,
            *,
            version: Strint = AUTO,
            python_home: str = None,
            enable_threads: bool = None,
            search_path: str = None,
            python_binary: str = None,
            tracebacker_path: str = None,
            plugin_dir: str = None,
            os_env_reload: bool = None,
            optimization_level: int = None
    ):
        """

        :param version: Python version plugin supports.

            Example:
                * 3 - version 3
                * <empty> - version 2
                * <default> - version deduced by uwsgiconf

        :param python_home: Set python executable directory - PYTHONHOME/virtualenv.

        :param enable_threads: Enable threads in the embedded languages.
            This will allow to spawn threads in your app.

            .. warning:: Threads will simply *not work* if this option is not enabled.
                         There will likely be no error, just no execution of your thread code.

        :param search_path: Add directory (or an .egg or a glob) to the Python search path.

            .. note:: This can be specified up to 64 times.

        :param python_binary: Set python program name.

        :param tracebacker_path: Enable the uWSGI Python tracebacker.
            http://uwsgi-docs.readthedocs.io/en/latest/Tracebacker.html

        :param plugin_dir: Directory to search for plugin.

        :param os_env_reload: Force ``os.environ`` reloading for every request.
            Used to allow setting of ``UWSGI_SETENV`` for Python applications.

        :param optimization_level: Python optimization level (see ``-O`` argument).
            .. warning:: This may be dangerous for some apps.

        """
        self._set_name(version)

        self._section.workers.set_thread_params(enable=enable_threads)
        self._set('py-tracebacker', tracebacker_path)
        self._set('py-program-name', python_binary)
        self._set('pyhome', python_home)
        self._set('pythonpath', search_path, multi=True)
        self._set('reload-os-env', os_env_reload, cast=bool)
        self._set('optimize', optimization_level)

        self._section.set_plugins_params(search_dirs=plugin_dir)

        return self._section

    def _set_name(self, version: Strint = AUTO):
        """Returns plugin name."""

        name = 'python'

        if version:
            if version is AUTO:
                version = sys.version_info[0]

                if version == 2:
                    version = ''

            name = f'{name}{version}'

        self.name = name

    def _get_name(self, *args, **kwargs):
        name = self.name

        if not name:
            self._set_name()

        return self.name

    def set_app_args(self, *args):
        """Sets ``sys.argv`` for python apps.

        Examples:
            * pyargv="one two three" will set ``sys.argv`` to ``('one', 'two', 'three')``.

        :param args:

        """
        if args:
            self._set('pyargv', ' '.join(args))

        return self._section

    def set_wsgi_params(self, *, module: Strpath = None, callable_name: str = None, env_strategy: str = None):
        """Set wsgi related parameters.

        :param module:
            * load .wsgi file as the Python application
            * load a WSGI module as the application.

            .. note:: The module (sans ``.py``) must be importable, i.e. be in ``PYTHONPATH``.

            Examples:
                * mypackage.my_wsgi_module -- read from `application` attr of mypackage/my_wsgi_module.py
                * mypackage.my_wsgi_module:my_app -- read from `my_app` attr of mypackage/my_wsgi_module.py

        :param callable_name: Set WSGI callable name. Default: application.

        :param env_strategy: Strategy for allocating/deallocating
            the WSGI env, can be:

            * ``cheat`` - preallocates the env dictionary on uWSGI startup and clears it
                after each request. Default behaviour for uWSGI <= 2.0.x

            * ``holy`` - creates and destroys the environ dictionary at each request.
                Default behaviour for uWSGI >= 2.1

        """
        module = str(module or '')

        if '/' in module:
            self._set('wsgi-file', module, condition=module)

        else:
            self._set('wsgi', module, condition=module)

        self._set('callable', callable_name)
        self._set('wsgi-env-behaviour', env_strategy)

        return self._section

    def eval_wsgi_entrypoint(self, code: str):
        """Evaluates Python code as WSGI entry point.

        :param code:

        """
        self._set('eval', code)

        return self._section

    def set_autoreload_params(self, *, scan_interval: int = None, ignore_modules: Strlist = None):
        """Sets autoreload related parameters.

        :param scan_interval: Seconds. Monitor Python modules' modification times to trigger reload.

            .. warning:: Use only in development.

        :param ignore_modules: Ignore the specified module during auto-reload scan.

        """
        self._set('py-auto-reload', scan_interval)
        self._set('py-auto-reload-ignore', ignore_modules, multi=True)

        return self._section

    def register_module_alias(self, alias: str, module_path: str, *, after_init: bool = False):
        """Adds an alias for a module.

        http://uwsgi-docs.readthedocs.io/en/latest/PythonModuleAlias.html

        :param alias:
        :param module_path:
        :param after_init: add a python module alias after uwsgi module initialization

        """
        command = 'post-pymodule-alias' if after_init else 'pymodule-alias'
        self._set(command, f'{alias}={module_path}', multi=True)

        return self._section

    def import_module(self, modules: Strint, *, shared: bool = False, into_spooler: bool = False):
        """Imports a python module.

        :param modules:

        :param shared: If shared import is done once in master process.
            Otherwise, import a python module in all the processes.
            This is done after fork but before request processing.

        :param into_spooler: Import a python module in the spooler.
            http://uwsgi-docs.readthedocs.io/en/latest/Spooler.html

        """
        if all((shared, into_spooler)):
            raise ConfigurationError('Unable to set both `shared` and `into_spooler` flags')

        if into_spooler:
            command = 'spooler-python-import'
        else:
            command = 'shared-python-import' if shared else 'python-import'

        self._set(command, modules, multi=True)

        return self._section

    def run_module(self, module: str):
        """Runs a Python script in the uWSGI environment.

        :param module:

        """
        self._set('pyrun', module)

        return self._section

1	import sys	3✔
2
3	from ..base import OptionsGroup	3✔
4	from ..exceptions import ConfigurationError	3✔
5	from ..typehints import Strint, Strlist, Strpath	3✔
6
7	AUTO = -1	3✔
8
9
10	class Python(OptionsGroup):	3✔
11	"""Python plugin options.
12
13	.. note:: By default the plugin does not initialize the GIL.
14	This means your app-generated threads will not run.
15	If you need threads, remember to enable them with ``enable_threads``.
16
17	"""
18	plugin = True	3✔
19
20	def set_basic_params(	3✔
21	self,
22	*,
23	version: Strint = AUTO,
24	python_home: str = None,
25	enable_threads: bool = None,
26	search_path: str = None,
27	python_binary: str = None,
28	tracebacker_path: str = None,
29	plugin_dir: str = None,
30	os_env_reload: bool = None,
31	optimization_level: int = None
32	):
33	"""
34
35	:param version: Python version plugin supports.
36
37	Example:
38	* 3 - version 3
39	* <empty> - version 2
40	* <default> - version deduced by uwsgiconf
41
42	:param python_home: Set python executable directory - PYTHONHOME/virtualenv.
43
44	:param enable_threads: Enable threads in the embedded languages.
45	This will allow to spawn threads in your app.
46
47	.. warning:: Threads will simply not work if this option is not enabled.
48	There will likely be no error, just no execution of your thread code.
49
50	:param search_path: Add directory (or an .egg or a glob) to the Python search path.
51
52	.. note:: This can be specified up to 64 times.
53
54	:param python_binary: Set python program name.
55
56	:param tracebacker_path: Enable the uWSGI Python tracebacker.
57	http://uwsgi-docs.readthedocs.io/en/latest/Tracebacker.html
58
59	:param plugin_dir: Directory to search for plugin.
60
61	:param os_env_reload: Force ``os.environ`` reloading for every request.
62	Used to allow setting of ``UWSGI_SETENV`` for Python applications.
63
64	:param optimization_level: Python optimization level (see ``-O`` argument).
65	.. warning:: This may be dangerous for some apps.
66
67	"""
68	self._set_name(version)	3✔
69
70	self._section.workers.set_thread_params(enable=enable_threads)	3✔
71	self._set('py-tracebacker', tracebacker_path)	3✔
72	self._set('py-program-name', python_binary)	3✔
73	self._set('pyhome', python_home)	3✔
74	self._set('pythonpath', search_path, multi=True)	3✔
75	self._set('reload-os-env', os_env_reload, cast=bool)	3✔
76	self._set('optimize', optimization_level)	3✔
77
78	self._section.set_plugins_params(search_dirs=plugin_dir)	3✔
79
80	return self._section	3✔
81
82	def _set_name(self, version: Strint = AUTO):	3✔
83	"""Returns plugin name."""
84
85	name = 'python'	3✔
86
87	if version:	3✔
88	if version is AUTO:	3✔
89	version = sys.version_info[0]	3✔
90
91	if version == 2:	3✔
UNCOV 92	version = ''	×
93
94	name = f'{name}{version}'	3✔
95
96	self.name = name	3✔
97
98	def _get_name(self, args, *kwargs):	3✔
99	name = self.name	3✔
100
101	if not name:	3✔
UNCOV 102	self._set_name()	×
103
104	return self.name	3✔
105
106	def set_app_args(self, *args):	3✔
107	"""Sets ``sys.argv`` for python apps.
108
109	Examples:
110	* pyargv="one two three" will set ``sys.argv`` to ``('one', 'two', 'three')``.
111
112	:param args:
113
114	"""
115	if args:	3✔
116	self._set('pyargv', ' '.join(args))	3✔
117
118	return self._section	3✔
119
120	def set_wsgi_params(self, *, module: Strpath = None, callable_name: str = None, env_strategy: str = None):	3✔
121	"""Set wsgi related parameters.
122
123	:param module:
124	* load .wsgi file as the Python application
125	* load a WSGI module as the application.
126
127	.. note:: The module (sans ``.py``) must be importable, i.e. be in ``PYTHONPATH``.
128
129	Examples:
130	* mypackage.my_wsgi_module -- read from `application` attr of mypackage/my_wsgi_module.py
131	* mypackage.my_wsgi_module:my_app -- read from `my_app` attr of mypackage/my_wsgi_module.py
132
133	:param callable_name: Set WSGI callable name. Default: application.
134
135	:param env_strategy: Strategy for allocating/deallocating
136	the WSGI env, can be:
137
138	* ``cheat`` - preallocates the env dictionary on uWSGI startup and clears it
139	after each request. Default behaviour for uWSGI <= 2.0.x
140
141	* ``holy`` - creates and destroys the environ dictionary at each request.
142	Default behaviour for uWSGI >= 2.1
143
144	"""
145	module = str(module or '')	3✔
146
147	if '/' in module:	3✔
148	self._set('wsgi-file', module, condition=module)	3✔
149
150	else:
151	self._set('wsgi', module, condition=module)	3✔
152
153	self._set('callable', callable_name)	3✔
154	self._set('wsgi-env-behaviour', env_strategy)	3✔
155
156	return self._section	3✔
157
158	def eval_wsgi_entrypoint(self, code: str):	3✔
159	"""Evaluates Python code as WSGI entry point.
160
161	:param code:
162
163	"""
164	self._set('eval', code)	3✔
165
166	return self._section	3✔
167
168	def set_autoreload_params(self, *, scan_interval: int = None, ignore_modules: Strlist = None):	3✔
169	"""Sets autoreload related parameters.
170
171	:param scan_interval: Seconds. Monitor Python modules' modification times to trigger reload.
172
173	.. warning:: Use only in development.
174
175	:param ignore_modules: Ignore the specified module during auto-reload scan.
176
177	"""
178	self._set('py-auto-reload', scan_interval)	3✔
179	self._set('py-auto-reload-ignore', ignore_modules, multi=True)	3✔
180
181	return self._section	3✔
182
183	def register_module_alias(self, alias: str, module_path: str, *, after_init: bool = False):	3✔
184	"""Adds an alias for a module.
185
186	http://uwsgi-docs.readthedocs.io/en/latest/PythonModuleAlias.html
187
188	:param alias:
189	:param module_path:
190	:param after_init: add a python module alias after uwsgi module initialization
191
192	"""
193	command = 'post-pymodule-alias' if after_init else 'pymodule-alias'	3✔
194	self._set(command, f'{alias}={module_path}', multi=True)	3✔
195
196	return self._section	3✔
197
198	def import_module(self, modules: Strint, *, shared: bool = False, into_spooler: bool = False):	3✔
199	"""Imports a python module.
200
201	:param modules:
202
203	:param shared: If shared import is done once in master process.
204	Otherwise, import a python module in all the processes.
205	This is done after fork but before request processing.
206
207	:param into_spooler: Import a python module in the spooler.
208	http://uwsgi-docs.readthedocs.io/en/latest/Spooler.html
209
210	"""
211	if all((shared, into_spooler)):	3✔
212	raise ConfigurationError('Unable to set both `shared` and `into_spooler` flags')	3✔
213
214	if into_spooler:	3✔
215	command = 'spooler-python-import'	3✔
216	else:
217	command = 'shared-python-import' if shared else 'python-import'	3✔
218
219	self._set(command, modules, multi=True)	3✔
220
221	return self._section	3✔
222
223	def run_module(self, module: str):	3✔
224	"""Runs a Python script in the uWSGI environment.
225
226	:param module:
227
228	"""
229	self._set('pyrun', module)	3✔
230
231	return self._section	3✔

idlesign / uwsgiconf / 13814784687

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