mirrors/granian
1
0
Fork 0
mirror of https://github.com/emmett-framework/granian.git synced 2025-08-04 17:08:02 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

Merge 8f117120b9 into 3f0bbd271d

Browse source
This commit is contained in:
Reda Bouaida 2025-07-06 16:35:13 +00:00 • committed by GitHub
commit 395e4db49f
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
3 changed files with 123 additions and 29 deletions
Download patch file Download diff file Expand all files Collapse all files

39
README.md
View file

@ -151,10 +151,10 @@ Options:
--blocking-threads INTEGER RANGE
Number of blocking threads (per worker)
[env var: GRANIAN_BLOCKING_THREADS; x>=1]
--blocking-threads-idle-timeout INTEGER RANGE
The maximum amount of time in seconds an
idle blocking thread will be kept alive
[env var:
--blocking-threads-idle-timeout DURATION
The maximum amount of time in seconds (or a
human-readable duration) an idle blocking
thread will be kept alive [env var:
GRANIAN_BLOCKING_THREADS_IDLE_TIMEOUT;
default: 30; 10<=x<=600]
--runtime-threads INTEGER RANGE
@ -221,8 +221,9 @@ Options:
connection alive [env var:
GRANIAN_HTTP2_KEEP_ALIVE_INTERVAL;
1<=x<=60000]
--http2-keep-alive-timeout INTEGER RANGE
Sets a timeout (in seconds) for receiving an
--http2-keep-alive-timeout DURATION
Sets a timeout (in seconds or a human-
readable duration) for receiving an
acknowledgement of the HTTP2 keep-alive ping
[env var: GRANIAN_HTTP2_KEEP_ALIVE_TIMEOUT;
default: 20; x>=1]
@ -277,16 +278,16 @@ Options:
--respawn-interval FLOAT The number of seconds to sleep between
workers respawn [env var:
GRANIAN_RESPAWN_INTERVAL; default: 3.5]
--workers-lifetime INTEGER RANGE
The maximum amount of time in seconds a
worker will be kept alive before respawn
[env var: GRANIAN_WORKERS_LIFETIME; x>=60]
--workers-kill-timeout INTEGER RANGE
The amount of time in seconds to wait for
killing workers that refused to gracefully
stop [env var:
GRANIAN_WORKERS_KILL_TIMEOUT; default:
(disabled); 1<=x<=1800]
--workers-lifetime DURATION The maximum amount of time in seconds (or a
human-readable duration) a worker will be
kept alive before respawn [env var:
GRANIAN_WORKERS_LIFETIME; x>=60]
--workers-kill-timeout DURATION
The amount of time in seconds (or a human-
readable duration) to wait for killing
workers that refused to gracefully stop
[env var: GRANIAN_WORKERS_KILL_TIMEOUT;
default: (disabled); 1<=x<=1800]
--factory / --no-factory Treat target as a factory function, that
should be invoked to build the actual target
[env var: GRANIAN_FACTORY; default:
@ -301,9 +302,9 @@ Options:
(/static)]
--static-path-mount DIRECTORY Path to mount for static file serving [env
var: GRANIAN_STATIC_PATH_MOUNT]
--static-path-expires INTEGER RANGE
Cache headers expiration (in seconds) for
static file serving [env var:
--static-path-expires DURATION Cache headers expiration (in seconds or a
human-readable duration) for static file
serving [env var:
GRANIAN_STATIC_PATH_EXPIRES; default: 86400;
x>=60]
--reload / --no-reload Enable auto reload on application's files

64
granian/cli.py
View file

@ -1,5 +1,6 @@
import json
import pathlib
import re
from enum import Enum
from typing import Any, Callable, List, Optional, Type, TypeVar, Union
@ -16,6 +17,49 @@ _AnyCallable = Callable[..., Any]
FC = TypeVar('FC', bound=Union[_AnyCallable, click.Command])
class Duration(click.IntRange):
"""Custom parameter type for duration strings like '24h', '6m', '2s', '1h30m', etc.
If the value is a plain number, it will be treated as seconds.
"""
name = 'duration'
_multipliers = {'s': 1, 'm': 60, 'h': 60 * 60, 'd': 60 * 60 * 24}
_pattern = re.compile(r'^(?:(?P<d>\d+)d)?(?:(?P<h>\d+)h)?(?:(?P<m>\d+)m)?(?:(?P<s>\d+)s)?$')
def __init__(self, min: Optional[int] = None, max: Optional[int] = None) -> None:
super().__init__(min, max, min_open=False, max_open=False, clamp=False)
def convert(self, value: Any, param: Optional[click.Parameter], ctx: Optional[click.Context]) -> Any:
if value is None:
return value
if isinstance(value, int):
seconds = value
elif isinstance(value, str):
if value.isnumeric():
seconds = int(value)
elif (match := self._pattern.fullmatch(value)) is not None:
seconds = (
int(match.group('d') or 0) * self._multipliers['d']
+ int(match.group('h') or 0) * self._multipliers['h']
+ int(match.group('m') or 0) * self._multipliers['m']
+ int(match.group('s') or 0) * self._multipliers['s']
)
else:
self.fail(f'{value!r} is not a valid duration', param, ctx)
else:
self.fail(f'{value!r} is not a valid duration', param, ctx)
if self.min is not None and seconds < self.min:
self.fail(f'{value!r} is less than the minimum allowed value of {self.min} seconds', param, ctx)
if self.max is not None and seconds > self.max:
self.fail(f'{value!r} is greater than the maximum allowed value of {self.max} seconds', param, ctx)
return seconds
class EnumType(click.Choice):
def __init__(self, enum: Enum, case_sensitive=False) -> None:
self.__enum = enum
@ -71,9 +115,9 @@ def option(*param_decls: str, cls: Optional[Type[click.Option]] = None, **attrs:
)
@option(
'--blocking-threads-idle-timeout',
type=click.IntRange(10, 600),
type=Duration(10, 600),
default=30,
help='The maximum amount of time in seconds an idle blocking thread will be kept alive',
help='The maximum amount of time in seconds (or a human-readable duration) an idle blocking thread will be kept alive',
)
@option('--runtime-threads', type=click.IntRange(1), default=1, help='Number of runtime threads (per worker)')
@option(
@ -153,9 +197,9 @@ def option(*param_decls: str, cls: Optional[Type[click.Option]] = None, **attrs:
)
@option(
'--http2-keep-alive-timeout',
type=click.IntRange(1),
type=Duration(1),
default=HTTP2Settings.keep_alive_timeout,
help='Sets a timeout (in seconds) for receiving an acknowledgement of the HTTP2 keep-alive ping',
help='Sets a timeout (in seconds or a human-readable duration) for receiving an acknowledgement of the HTTP2 keep-alive ping',
)
@option(
'--http2-max-concurrent-streams',
@ -230,13 +274,13 @@ def option(*param_decls: str, cls: Optional[Type[click.Option]] = None, **attrs:
)
@option(
'--workers-lifetime',
type=click.IntRange(60),
help='The maximum amount of time in seconds a worker will be kept alive before respawn',
type=Duration(60),
help='The maximum amount of time in seconds (or a human-readable duration) a worker will be kept alive before respawn',
)
@option(
'--workers-kill-timeout',
type=click.IntRange(1, 1800),
help='The amount of time in seconds to wait for killing workers that refused to gracefully stop',
type=Duration(1, 1800),
help='The amount of time in seconds (or a human-readable duration) to wait for killing workers that refused to gracefully stop',
show_default='disabled',
)
@option(
@ -267,9 +311,9 @@ def option(*param_decls: str, cls: Optional[Type[click.Option]] = None, **attrs:
)
@option(
'--static-path-expires',
type=click.IntRange(60),
type=Duration(60),
default=86400,
help='Cache headers expiration (in seconds) for static file serving',
help='Cache headers expiration (in seconds or a human-readable duration) for static file serving',
)
@option(
'--reload/--no-reload',

49
tests/test_cli.py Normal file
View file

@ -0,0 +1,49 @@
import click
import pytest
from granian.cli import Duration
@pytest.mark.parametrize(
('value', 'expected'),
(
('10', 10),
('10s', 10),
('10m', 60 * 10),
('10m10s', 60 * 10 + 10),
('10h', 60 * 60 * 10),
('10d', 24 * 60 * 60 * 10),
('10d10h10m10s', 24 * 60 * 60 * 10 + 60 * 60 * 10 + 60 * 10 + 10),
),
)
def test_duration_convert(value: str, expected: int) -> None:
duration_type = Duration()
assert duration_type.convert(value, None, None) == expected
@pytest.mark.parametrize(
('value', 'error_message'),
(
('10x', r"'10x' is not a valid duration"),
('10d10h10m10s10', r"'10d10h10m10s10' is not a valid duration"),
('10d10h10m10s10', r"'10d10h10m10s10' is not a valid duration"),
),
)
def test_duration_convert_invalid(value: str, error_message: str) -> None:
duration_type = Duration()
with pytest.raises(click.BadParameter, match=error_message):
duration_type.convert(value, None, None)
@pytest.mark.parametrize(
('value', 'error_message'),
(
('1000', r"'1000' is greater than the maximum allowed value of 100 seconds"),
('30m', r"'30m' is greater than the maximum allowed value of 100 seconds"),
('5', r"'5' is less than the minimum allowed value of 10 seconds"),
),
)
def test_duration_convert_out_of_range(value: str, error_message: str) -> None:
duration_type = Duration(10, 100)
with pytest.raises(click.BadParameter, match=error_message):
duration_type.convert(value, None, None)