22310118381

Committed 23 Feb 2026 02:22PM UTC coverage: 95.817% (+0.01%) from 95.807%

Build # 22310118381

Build Type

push

github

Committed by

pylover

Commit Message

v7.3.0

Coverage Stats

1 of 1 new or added line in 1 file covered. (100.0%)

1 existing line in 1 file now uncovered.

756 of 789 relevant lines covered (95.82%)

2.87 hits per line

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

99.3

/yhttp/core/application.py

import inspect
import functools
import re
import types

import pymlconf

from . import statuses, static
from .request import Request
from .response import Response
from .cli import Main
from .guard import Guard
from .multidict import MultiDict


class BaseApplication:
    """Base application for :class:`Application` and :class:`Rewrite`

    :param version: Application version
    :param name: Application name

    .. versionadded:: 6.4

       ``name`` argument.

    .. versionchanged:: 7.0

       ``version`` and ``name`` arguments must be provided to create an
       Application.

    """
    _builtinsettings = '''
    debug: true
    '''

    #: Instance of :class:`pymlconf.Root` as the global configuration instance.
    settings = None

    #: A list of :class:`easycli.Argument` or :class:`easycli.SubCommand`.
    cliarguments = None

    #: A dictionary to hold registered functions to specific hooks.
    events = None

    def __init__(self, version, name):
        self.version = version
        self.name = name
        self.events = {}
        self.cliarguments = []
        self.settings = pymlconf.Root(self._builtinsettings)

    def when(self, func):
        """Return decorator to registers the ``func`` into :attr:`events` by
        its name.

        Currently these hooks are suuported:

        * ready
        * shutdown
        * startresponse
        * endresponse

        .. versionadded:: 7.3

           ``startresponse`` hook.

        The hook name will be choosed by the func.__name__, so if you need to
        aware when :meth:`ready` is called write something like this:

        .. code-block::

           @app.when
           def ready(app):
               ...

           @app.when
           def shutdown(app):
               ...

           @app.when
           def startresponse(response):
               ...

           @app.when
           def endresponse(response):
               ...

        """
        callbacks = self.events.setdefault(func.__name__, [])
        if func not in callbacks:
            callbacks.append(func)

    def hook(self, name, *a, **kw):
        """Only way to fire registered hooks.

        Hooks can registered by :meth:`when()` with the name.

        .. code-block::

           app.hook('endresponse')

        Extra parameters: ``*a, **kw`` will be passed to event handlers.

        Normally, users no need to call this method.
        """
        callbacks = self.events.get(name)
        if not callbacks:
            return

        for c in callbacks:
            c(*a, **kw)

    def ready(self):
        """Call the ``ready`` :meth:`hook`.

        You need to call this method before using the instance as the WSGI
        application.

        Typical usage:

        .. code-block::

           from yhttp.core import Application, text


           app = Application()

           @app.route()
           @text
           def get(req):
               return 'Hello World!'

           if __name__ != '__main__':
               app.ready()
        """
        self.hook('ready', self)

    def shutdown(self):
        """Call the ``shutdown`` :meth:`hook`.
        """
        self.hook('shutdown', self)

    def climain(self, argv=None):
        """Provide a callable to call as the CLI entry point.

        .. code-block::

           import sys


           if __name__ == '__main__':
               sys.exit(app.climain(sys.argv))

        You can use this method as the setuptools entry point for
        `Automatic Script Creation <https://setuptools.readthedocs.io/en/la\
        test/setuptools.html#automatic-script-creation>`_

        ``setup.py``

        .. code-block::

           from setuptools import setup


           setup(
               name='foo',
               ...
               entry_points={
                   'console_scripts': [
                       'foo = foo:app.climain'
                   ]
               }
           )

        .. seealso::

           :ref:`quickstart-commandlineinterface`

        """
        return Main(self).main(argv)


class Application(BaseApplication):
    """WSGI Web Application.

    Instance of this class can be used as a WSGI application.

    :cvar bodyguard_factory: A factory of :class:`.Guard` and or it's
                             subclasses to be used in
                             :meth:`Application.bodyguard` to instantiate a new
                             guard for a handler. default: :class:`.Guard`.
    :cvar queryguard_factory: A factory of :class:`.Guard` and or it's
                              subclasses to be used in
                              :meth:`Application.queryguard` to instantiate a
                              new guard for a handler.
                              default: :class:`.Guard`.
    :ivar routes: A dictionionary to hold the regext routes handler mappings.

    :param version: Application version
    :param name: Application name
    """

    _builtinsettings = '''
    debug: true
    staticdir:
        autoindex: true
        default: index.html
        fallback: index.html
    '''

    routes = None
    bodyguard_factory = Guard
    queryguard_factory = Guard

    def __init__(self, version, name):
        self.routes = {}
        super().__init__(version=version, name=name)

    def _matchrequest(self, patterns, request):
        for pattern, handler, info in patterns:
            match = pattern.match(request.path)
            if not match:
                continue

            pathparams = [a for a in match.groups() if a is not None]
            query = {
                k: v for k, v in request.query.items()
                if k in info['kwonly']
            }

            return handler, pathparams, query

        return None, None, None

    def _findhandler(self, request):
        # All verbs
        patterns = self.routes.get('*', [])
        if patterns:
            handler, args, query = self._matchrequest(patterns, request)
            if handler is not None:
                return handler, args, query

        # Specific verb
        patterns = self.routes.get(request.verb.upper())
        if not patterns:
            raise statuses.methodnotallowed()

        handler, args, query = self._matchrequest(patterns, request)
        if handler is None:
            raise statuses.notfound()

        return handler, args, query

    def __call__(self, environ, startresponse):
        """Actual WSGI Application.

        So, will be called on every request.

        .. code-block::

           from yhttp.core import Application


           app = Application()
           result = app(environ, start_response)

        Checkout the `PEP 333 <https://www.python.org/dev/peps/pep-0333/>`_
        for more info.

        """
        response = Response(self, environ, startresponse)
        request = Request(self, environ, response)

        try:
            handler, pathparams, query = self._findhandler(request)
            body = handler(request, *pathparams, **query)

            if isinstance(body, statuses.HTTPStatus):
                raise body

            if isinstance(body, types.GeneratorType):
                response._firstchunk = next(body)

            response.body = body

        except statuses.HTTPStatus as ex:
            ex.setupresponse(response, stacktrace=self.settings.debug)

        return response.start()

    def delete_route(self, pattern, verb, flags=0):
        r"""Delete a route

        :param pattern: Regular expression to match the routing table.
        :param flags: Regular expression flags. see :func:`re.compile`.
        :param verb: The HTTP verb to match the routing table.
        """
        routes = self.routes[verb.upper()]

        pat = re.compile(f'^{pattern}$', flags)
        for r in routes:
            if r[0] == pat:
                routes.remove(r)
                return

        raise ValueError(f'Route not exists: {pattern}')

    def route(self, pattern='/', flags=0, verb=None, insert=None,
              exists='error'):
        r"""Return a decorator to register a handler for given regex pattern.

        if ``verb`` is ``None`` then the function name will used instead.


        .. code-block::

           @app.route(r'/.*')
           def get(req):
               ...

        You can bypass this behavior by passing ``verb`` keyword argument:

        .. code-block::

           @app.route(r'/', verb='get')
           def somethingelse(req):
               ...

        To catch any verb by the handler use ``*``.

        .. code-block::

           @app.route(r'/', verb='*')
           def any(req):
               ...

        Regular expression groups will be capture and dispatched as the
        positional arguments of the handler after ``req``:

        .. code-block::

           @app.route(r'/(\\d+)/(\\w*)')
           def get(req, id, name):
               ...

        This method returns a decorator for handler fucntions. So, you can use
        it like:

        .. code-block::

           books = app.route(r'/books/(.*)')

           @books
           def get(req, id):
               ...

           @books
           def post(req, id):
               ...

        .. seealso::

           :ref:`cookbook-routing`
           :ref:`cookbook-pathparams`


        :param pattern: Regular expression to match the request.
        :param flags: Regular expression flags. see :func:`re.compile`.
        :param verb: If not given then ``handler.__name__`` will be used to
                     match HTYP verb, Use ``*`` to catch all verbs.
        :param insert: If not given, route will be appended to the end of the
                       :attr:`routes`. Otherwise it must be an
                       integer indicating the place to insert the new route
                       into :attr:`routes` attribute.
        :param exists: Tell what to do if route already exists, possible
                       values: ``error``(default) and ``remove`` to remove the
                       existing route before appending and or inserting the new
                       one.

        .. versionadded:: 2.9

           ``insert``

        .. versionadded:: 6.1

           ``exists``

        """
        if exists not in ('error', 'remove'):
            raise ValueError('Invalid value for exists argument, use one of '
                             '`error` (the default) and or `remove`.')

        def decorator(handler):
            nonlocal verb, exists
            verbs = verb or handler.__name__

            if isinstance(verbs, str):
                verbs = [verbs]

            for verb in verbs:
                routes = self.routes.setdefault(verb.upper(), [])
                signature = inspect.signature(handler)
                info = dict(
                    kwonly={
                        k for k, v in signature.parameters.items()
                        if v.kind == inspect.Parameter.KEYWORD_ONLY
                    }
                )
                pat = re.compile(f'^{pattern}$', flags)
                for r in routes:
                    if r[0] == pat:
                        if exists == 'error':
                            raise ValueError(
                                f'Route already exists: {pattern}')

                        routes.remove(r)
                        break

                route = (pat, handler, info)
                if insert is not None:
                    routes.insert(insert, route)
                else:
                    routes.append(route)

        return decorator

    def staticfile(self, pattern, filename, **kw):
        r"""Register a filename with a regular expression pattern to be served.

        .. code-block::

            app.staticfile(r'/a\.txt', 'physical/path/to/a.txt')

        .. seealso::

           :ref:`cookbook-static`

        """
        return self.route(pattern, **kw)(static.file(filename))

    def staticdirectory(self, pattern, directory, default=None, autoindex=True,
                        fallback=None, **kw):
        r"""Register a directory with a regular expression pattern.

        So the files inside the directory are accessible by their names:

        .. code-block::

            app.staticdirectory(r'/foo/', 'physical/path/to/foo')

        You you can do:

        .. code-block:: bash

           curl localhost:8080/foo/a.txt

        .. seealso::

           :ref:`cookbook-static`

        :param pattern: Regular expression to match the requests.
        :param directory: Static files are here.
        :param default: if None, the ``app.settings.staticdir.default``
                        (which default is ``index.html``) will be used as the
                        default document.
        :param autoindex: Automatic directory indexing, default True.
        :param fallback: if ``True``, the ``app.settings.staticdir.fallback``
                        (which default is ``index.html``) will be used as the
                        fallback document if the requested resource was not
                        found. if ``str``, the value will be used instead of
                        ``app.settings.staticdir.fallback``.

        .. versionadded:: 2.13

           The *default* and *fallback* keyword arguments.

        .. versionadded:: 3.8

           The *autoindex* keyword argument.

        """
        return self.route(f'{pattern}(.*)', **kw)(static.directory(
            directory,
            default,
            autoindex,
            fallback
        ))

    def bodyguard(self, fields=None, strict=False):
        r"""A decorator factory to validate HTTP request's body.

        .. versionadded:: 5.1

        .. code-block::

           from yhttp.core import guard as g

           @app.route()
           @app.bodyguard(fields=(
               g.String('foo', length=(1, 8), pattern=r'\d+', optional=True),
               g.Integer('bar', range=(0, 9), optional=True),
           ), strict=True)
           @json()
           def post(req):
               ...

        This method calls the :attr:`bodyguard_factory` to
        intantiate a :class:`Guard` class or it's subclasses.

        :param fields: A tuple of :class:`Gurad.Field` subclass instances to
                       define the allowed fields and field attributes.
        :param strict: If ``True``, it raises
                       :attr:`Guard.statuscode_unknownfields` when one or more
                       fields are not in the given ``fields`` argument.
        """
        guard = self.bodyguard_factory(fields, strict)

        def decorator(handler):
            @functools.wraps(handler)
            def _handler(req, *args, **kwargs):
                if strict and (not fields) and req.contentlength:
                    # Body not allowed
                    raise statuses.badrequest()

                req.form = guard.validate(
                    req,
                    req.getform(relax=True) or MultiDict()
                )
                return handler(req, *args, **kwargs)

            return _handler

        return decorator

    def queryguard(self, fields=None, strict=False):
        """A decorator factory to validate the URL's query string.

        .. versionadded:: 5.1

        .. code-block::

           from yhttp.core import guard as g
           from yhttp.core.multidict import MultiDict

           def bar(req, field: g.Field, values: MultiDict):
               return 'bar default value'

           @app.route()
           @app.queryguard(fields=(
               g.String(
                   'foo',
                   length=(1, 8),
                   pattern=r'\\d+',
                   optional=True,
                   default='foo default value',
               ),
               g.Integer(
                   'bar',
                   range=(0, 9),
                   optional=True,
                   default=bar
              ),
           ), strict=True)
           @json()
           def post(req):
               ...

        This method calls the :attr:`queryguard_factory` to
        intantiate a :class:`Guard` class or it's subclasses.

        :param fields: A tuple of :class:`Gurad.Field` subclass instances to
                       define the allowed fields and field attributes.
        :param strict: If ``True``, it raises
                       :attr:`Guard.statuscode_unknownfields` when one or more
                       fields are not in the given ``fields`` argument.
        """
        guard = self.queryguard_factory(fields, strict)

        def decorator(handler):
            @functools.wraps(handler)
            def _handler(req, *args, **kwargs):
                if strict and (not fields) and req.query:
                    # Body not allowed
                    raise statuses.badrequest()

                req.query = guard.validate(req, req.query or MultiDict())
                for k in req.query:
                    if k in kwargs:
                        kwargs[k] = req.query[k]
                return handler(req, *args, **kwargs)

            return _handler

        return decorator

1	import inspect	3✔
2	import functools	3✔
3	import re	3✔
4	import types	3✔
5
6	import pymlconf	3✔
7
8	from . import statuses, static	3✔
9	from .request import Request	3✔
10	from .response import Response	3✔
11	from .cli import Main	3✔
12	from .guard import Guard	3✔
13	from .multidict import MultiDict	3✔
14
15
16	class BaseApplication:	3✔
17	"""Base application for :class:`Application` and :class:`Rewrite`
18
19	:param version: Application version
20	:param name: Application name
21
22	.. versionadded:: 6.4
23
24	``name`` argument.
25
26	.. versionchanged:: 7.0
27
28	``version`` and ``name`` arguments must be provided to create an
29	Application.
30
31	"""
32	_builtinsettings = '''	3✔
33	debug: true
34	'''
35
36	#: Instance of :class:`pymlconf.Root` as the global configuration instance.
37	settings = None	3✔
38
39	#: A list of :class:`easycli.Argument` or :class:`easycli.SubCommand`.
40	cliarguments = None	3✔
41
42	#: A dictionary to hold registered functions to specific hooks.
43	events = None	3✔
44
45	def __init__(self, version, name):	3✔
46	self.version = version	3✔
47	self.name = name	3✔
48	self.events = {}	3✔
49	self.cliarguments = []	3✔
50	self.settings = pymlconf.Root(self._builtinsettings)	3✔
51
52	def when(self, func):	3✔
53	"""Return decorator to registers the ``func`` into :attr:`events` by
54	its name.
55
56	Currently these hooks are suuported:
57
58	* ready
59	* shutdown
60	* startresponse
61	* endresponse
62
63	.. versionadded:: 7.3
64
65	``startresponse`` hook.
66
67	The hook name will be choosed by the func.__name__, so if you need to
68	aware when :meth:`ready` is called write something like this:
69
70	.. code-block::
71
72	@app.when
73	def ready(app):
74	...
75
76	@app.when
77	def shutdown(app):
78	...
79
80	@app.when
81	def startresponse(response):
82	...
83
84	@app.when
85	def endresponse(response):
86	...
87
88	"""
89	callbacks = self.events.setdefault(func.__name__, [])	3✔
90	if func not in callbacks:	3✔
91	callbacks.append(func)	3✔
92
93	def hook(self, name, a, *kw):	3✔
94	"""Only way to fire registered hooks.
95
96	Hooks can registered by :meth:`when()` with the name.
97
98	.. code-block::
99
100	app.hook('endresponse')
101
102	Extra parameters: ``a, *kw`` will be passed to event handlers.
103
104	Normally, users no need to call this method.
105	"""
106	callbacks = self.events.get(name)	3✔
107	if not callbacks:	3✔
108	return	3✔
109
110	for c in callbacks:	3✔
111	c(a, *kw)	3✔
112
113	def ready(self):	3✔
114	"""Call the ``ready`` :meth:`hook`.
115
116	You need to call this method before using the instance as the WSGI
117	application.
118
119	Typical usage:
120
121	.. code-block::
122
123	from yhttp.core import Application, text
124
125
126	app = Application()
127
128	@app.route()
129	@text
130	def get(req):
131	return 'Hello World!'
132
133	if __name__ != '__main__':
134	app.ready()
135	"""
136	self.hook('ready', self)	3✔
137
138	def shutdown(self):	3✔
139	"""Call the ``shutdown`` :meth:`hook`.
140	"""
141	self.hook('shutdown', self)	3✔
142
143	def climain(self, argv=None):	3✔
144	"""Provide a callable to call as the CLI entry point.
145
146	.. code-block::
147
148	import sys
149
150
151	if __name__ == '__main__':
152	sys.exit(app.climain(sys.argv))
153
154	You can use this method as the setuptools entry point for
155	`Automatic Script Creation <https://setuptools.readthedocs.io/en/la\
156	test/setuptools.html#automatic-script-creation>`_
157
158	``setup.py``
159
160	.. code-block::
161
162	from setuptools import setup
163
164
165	setup(
166	name='foo',
167	...
168	entry_points={
169	'console_scripts': [
170	'foo = foo:app.climain'
171	]
172	}
173	)
174
175	.. seealso::
176
177	:ref:`quickstart-commandlineinterface`
178
179	"""
UNCOV 180	return Main(self).main(argv)	×
181
182
183	class Application(BaseApplication):	3✔
184	"""WSGI Web Application.
185
186	Instance of this class can be used as a WSGI application.
187
188	:cvar bodyguard_factory: A factory of :class:`.Guard` and or it's
189	subclasses to be used in
190	:meth:`Application.bodyguard` to instantiate a new
191	guard for a handler. default: :class:`.Guard`.
192	:cvar queryguard_factory: A factory of :class:`.Guard` and or it's
193	subclasses to be used in
194	:meth:`Application.queryguard` to instantiate a
195	new guard for a handler.
196	default: :class:`.Guard`.
197	:ivar routes: A dictionionary to hold the regext routes handler mappings.
198
199	:param version: Application version
200	:param name: Application name
201	"""
202
203	_builtinsettings = '''	3✔
204	debug: true
205	staticdir:
206	autoindex: true
207	default: index.html
208	fallback: index.html
209	'''
210
211	routes = None	3✔
212	bodyguard_factory = Guard	3✔
213	queryguard_factory = Guard	3✔
214
215	def __init__(self, version, name):	3✔
216	self.routes = {}	3✔
217	super().__init__(version=version, name=name)	3✔
218
219	def _matchrequest(self, patterns, request):	3✔
220	for pattern, handler, info in patterns:	3✔
221	match = pattern.match(request.path)	3✔
222	if not match:	3✔
223	continue	3✔
224
225	pathparams = [a for a in match.groups() if a is not None]	3✔
226	query = {	3✔
227	k: v for k, v in request.query.items()
228	if k in info['kwonly']
229	}
230
231	return handler, pathparams, query	3✔
232
233	return None, None, None	3✔
234
235	def _findhandler(self, request):	3✔
236	# All verbs
237	patterns = self.routes.get('*', [])	3✔
238	if patterns:	3✔
239	handler, args, query = self._matchrequest(patterns, request)	3✔
240	if handler is not None:	3✔
241	return handler, args, query	3✔
242
243	# Specific verb
244	patterns = self.routes.get(request.verb.upper())	3✔
245	if not patterns:	3✔
246	raise statuses.methodnotallowed()	3✔
247
248	handler, args, query = self._matchrequest(patterns, request)	3✔
249	if handler is None:	3✔
250	raise statuses.notfound()	3✔
251
252	return handler, args, query	3✔
253
254	def __call__(self, environ, startresponse):	3✔
255	"""Actual WSGI Application.
256
257	So, will be called on every request.
258
259	.. code-block::
260
261	from yhttp.core import Application
262
263
264	app = Application()
265	result = app(environ, start_response)
266
267	Checkout the `PEP 333 <https://www.python.org/dev/peps/pep-0333/>`_
268	for more info.
269
270	"""
271	response = Response(self, environ, startresponse)	3✔
272	request = Request(self, environ, response)	3✔
273
274	try:	3✔
275	handler, pathparams, query = self._findhandler(request)	3✔
276	body = handler(request, pathparams, *query)	3✔
277
278	if isinstance(body, statuses.HTTPStatus):	3✔
279	raise body	3✔
280
281	if isinstance(body, types.GeneratorType):	3✔
282	response._firstchunk = next(body)	3✔
283
284	response.body = body	3✔
285
286	except statuses.HTTPStatus as ex:	3✔
287	ex.setupresponse(response, stacktrace=self.settings.debug)	3✔
288
289	return response.start()	3✔
290
291	def delete_route(self, pattern, verb, flags=0):	3✔
292	r"""Delete a route
293
294	:param pattern: Regular expression to match the routing table.
295	:param flags: Regular expression flags. see :func:`re.compile`.
296	:param verb: The HTTP verb to match the routing table.
297	"""
298	routes = self.routes[verb.upper()]	3✔
299
300	pat = re.compile(f'^{pattern}$', flags)	3✔
301	for r in routes:	3✔
302	if r[0] == pat:	3✔
303	routes.remove(r)	3✔
304	return	3✔
305
306	raise ValueError(f'Route not exists: {pattern}')	3✔
307
308	def route(self, pattern='/', flags=0, verb=None, insert=None,	3✔
309	exists='error'):
310	r"""Return a decorator to register a handler for given regex pattern.
311
312	if ``verb`` is ``None`` then the function name will used instead.
313
314
315	.. code-block::
316
317	@app.route(r'/.*')
318	def get(req):
319	...
320
321	You can bypass this behavior by passing ``verb`` keyword argument:
322
323	.. code-block::
324
325	@app.route(r'/', verb='get')
326	def somethingelse(req):
327	...
328
329	To catch any verb by the handler use ``*``.
330
331	.. code-block::
332
333	@app.route(r'/', verb='*')
334	def any(req):
335	...
336
337	Regular expression groups will be capture and dispatched as the
338	positional arguments of the handler after ``req``:
339
340	.. code-block::
341
342	@app.route(r'/(\\d+)/(\\w*)')
343	def get(req, id, name):
344	...
345
346	This method returns a decorator for handler fucntions. So, you can use
347	it like:
348
349	.. code-block::
350
351	books = app.route(r'/books/(.*)')
352
353	@books
354	def get(req, id):
355	...
356
357	@books
358	def post(req, id):
359	...
360
361	.. seealso::
362
363	:ref:`cookbook-routing`
364	:ref:`cookbook-pathparams`
365
366
367	:param pattern: Regular expression to match the request.
368	:param flags: Regular expression flags. see :func:`re.compile`.
369	:param verb: If not given then ``handler.__name__`` will be used to
370	match HTYP verb, Use ``*`` to catch all verbs.
371	:param insert: If not given, route will be appended to the end of the
372	:attr:`routes`. Otherwise it must be an
373	integer indicating the place to insert the new route
374	into :attr:`routes` attribute.
375	:param exists: Tell what to do if route already exists, possible
376	values: ``error``(default) and ``remove`` to remove the
377	existing route before appending and or inserting the new
378	one.
379
380	.. versionadded:: 2.9
381
382	``insert``
383
384	.. versionadded:: 6.1
385
386	``exists``
387
388	"""
389	if exists not in ('error', 'remove'):	3✔
390	raise ValueError('Invalid value for exists argument, use one of '	3✔
391	'`error` (the default) and or `remove`.')
392
393	def decorator(handler):	3✔
394	nonlocal verb, exists
395	verbs = verb or handler.__name__	3✔
396
397	if isinstance(verbs, str):	3✔
398	verbs = [verbs]	3✔
399
400	for verb in verbs:	3✔
401	routes = self.routes.setdefault(verb.upper(), [])	3✔
402	signature = inspect.signature(handler)	3✔
403	info = dict(	3✔
404	kwonly={
405	k for k, v in signature.parameters.items()
406	if v.kind == inspect.Parameter.KEYWORD_ONLY
407	}
408	)
409	pat = re.compile(f'^{pattern}$', flags)	3✔
410	for r in routes:	3✔
411	if r[0] == pat:	3✔
412	if exists == 'error':	3✔
413	raise ValueError(	3✔
414	f'Route already exists: {pattern}')
415
416	routes.remove(r)	3✔
417	break	3✔
418
419	route = (pat, handler, info)	3✔
420	if insert is not None:	3✔
421	routes.insert(insert, route)	3✔
422	else:
423	routes.append(route)	3✔
424
425	return decorator	3✔
426
427	def staticfile(self, pattern, filename, **kw):	3✔
428	r"""Register a filename with a regular expression pattern to be served.
429
430	.. code-block::
431
432	app.staticfile(r'/a\.txt', 'physical/path/to/a.txt')
433
434	.. seealso::
435
436	:ref:`cookbook-static`
437
438	"""
439	return self.route(pattern, **kw)(static.file(filename))	3✔
440
441	def staticdirectory(self, pattern, directory, default=None, autoindex=True,	3✔
442	fallback=None, **kw):
443	r"""Register a directory with a regular expression pattern.
444
445	So the files inside the directory are accessible by their names:
446
447	.. code-block::
448
449	app.staticdirectory(r'/foo/', 'physical/path/to/foo')
450
451	You you can do:
452
453	.. code-block:: bash
454
455	curl localhost:8080/foo/a.txt
456
457	.. seealso::
458
459	:ref:`cookbook-static`
460
461	:param pattern: Regular expression to match the requests.
462	:param directory: Static files are here.
463	:param default: if None, the ``app.settings.staticdir.default``
464	(which default is ``index.html``) will be used as the
465	default document.
466	:param autoindex: Automatic directory indexing, default True.
467	:param fallback: if ``True``, the ``app.settings.staticdir.fallback``
468	(which default is ``index.html``) will be used as the
469	fallback document if the requested resource was not
470	found. if ``str``, the value will be used instead of
471	``app.settings.staticdir.fallback``.
472
473	.. versionadded:: 2.13
474
475	The default and fallback keyword arguments.
476
477	.. versionadded:: 3.8
478
479	The autoindex keyword argument.
480
481	"""
482	return self.route(f'{pattern}(.)', *kw)(static.directory(	3✔
483	directory,
484	default,
485	autoindex,
486	fallback
487	))
488
489	def bodyguard(self, fields=None, strict=False):	3✔
490	r"""A decorator factory to validate HTTP request's body.
491
492	.. versionadded:: 5.1
493
494	.. code-block::
495
496	from yhttp.core import guard as g
497
498	@app.route()
499	@app.bodyguard(fields=(
500	g.String('foo', length=(1, 8), pattern=r'\d+', optional=True),
501	g.Integer('bar', range=(0, 9), optional=True),
502	), strict=True)
503	@json()
504	def post(req):
505	...
506
507	This method calls the :attr:`bodyguard_factory` to
508	intantiate a :class:`Guard` class or it's subclasses.
509
510	:param fields: A tuple of :class:`Gurad.Field` subclass instances to
511	define the allowed fields and field attributes.
512	:param strict: If ``True``, it raises
513	:attr:`Guard.statuscode_unknownfields` when one or more
514	fields are not in the given ``fields`` argument.
515	"""
516	guard = self.bodyguard_factory(fields, strict)	3✔
517
518	def decorator(handler):	3✔
519	@functools.wraps(handler)	3✔
520	def _handler(req, args, *kwargs):	3✔
521	if strict and (not fields) and req.contentlength:	3✔
522	# Body not allowed
523	raise statuses.badrequest()	3✔
524
525	req.form = guard.validate(	3✔
526	req,
527	req.getform(relax=True) or MultiDict()
528	)
529	return handler(req, args, *kwargs)	3✔
530
531	return _handler	3✔
532
533	return decorator	3✔
534
535	def queryguard(self, fields=None, strict=False):	3✔
536	"""A decorator factory to validate the URL's query string.
537
538	.. versionadded:: 5.1
539
540	.. code-block::
541
542	from yhttp.core import guard as g
543	from yhttp.core.multidict import MultiDict
544
545	def bar(req, field: g.Field, values: MultiDict):
546	return 'bar default value'
547
548	@app.route()
549	@app.queryguard(fields=(
550	g.String(
551	'foo',
552	length=(1, 8),
553	pattern=r'\\d+',
554	optional=True,
555	default='foo default value',
556	),
557	g.Integer(
558	'bar',
559	range=(0, 9),
560	optional=True,
561	default=bar
562	),
563	), strict=True)
564	@json()
565	def post(req):
566	...
567
568	This method calls the :attr:`queryguard_factory` to
569	intantiate a :class:`Guard` class or it's subclasses.
570
571	:param fields: A tuple of :class:`Gurad.Field` subclass instances to
572	define the allowed fields and field attributes.
573	:param strict: If ``True``, it raises
574	:attr:`Guard.statuscode_unknownfields` when one or more
575	fields are not in the given ``fields`` argument.
576	"""
577	guard = self.queryguard_factory(fields, strict)	3✔
578
579	def decorator(handler):	3✔
580	@functools.wraps(handler)	3✔
581	def _handler(req, args, *kwargs):	3✔
582	if strict and (not fields) and req.query:	3✔
583	# Body not allowed
584	raise statuses.badrequest()	3✔
585
586	req.query = guard.validate(req, req.query or MultiDict())	3✔
587	for k in req.query:	3✔
588	if k in kwargs:	3✔
589	kwargs[k] = req.query[k]	3✔
590	return handler(req, args, *kwargs)	3✔
591
592	return _handler	3✔
593
594	return decorator	3✔

yhttp / yhttp / 22310118381

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