def get_pagination_options(params, default):
try:
opts = voluptuous.Schema({
voluptuous.Required(
"limit", default=pecan.request.conf.api.max_limit):
voluptuous.All(voluptuous.Coerce(int),
voluptuous.Range(min=1),
voluptuous.Clamp(
min=1, max=pecan.request.conf.api.max_limit)),
"marker": six.text_type,
voluptuous.Required("sort", default=default):
voluptuous.All(
voluptuous.Coerce(arg_to_list),
[six.text_type]),
}, extra=voluptuous.REMOVE_EXTRA)(params)
except voluptuous.Invalid as e:
abort(400, {"cause": "Argument value error",
"reason": str(e)})
opts['sorts'] = opts['sort']
del opts['sort']
return opts
python类REMOVE_EXTRA的实例源码
def __init__(self, server):
self.server = server
self.user_coll = server.user_coll
self.app_coll = server.app_coll
self.apps = server.apps
self.login_schema = Schema({
'email': str,
'password': str,
}, extra=REMOVE_EXTRA)
self.useradd_schema = Schema({
'email': str,
'password': str,
'username': str,
}, extra=REMOVE_EXTRA)
_o = Optional
self.app_add_schema = Schema({
'name': str,
_o('description'): str,
_o('bot_public'): bool,
_o('icon'): str,
}, extra=REMOVE_EXTRA)
self.register()
def __init__(self, server):
self.server = server
self.guild_man = server.guild_man
_o = Optional
self.guild_edit_schema = Schema({
_o('name'): str,
_o('region'): str,
_o('verification_level'): int,
_o('default_message_notifications'): int,
_o('afk_channel_id'): str,
_o('afk_timeout'): int,
_o('icon'): str,
_o('owner_id'): str,
}, required=True, extra=REMOVE_EXTRA)
self.guild_create_schema = Schema({
'name': str,
'region': str,
'icon': Any(None, str),
'verification_level': int,
'default_message_notifications': int,
}, extra=REMOVE_EXTRA)
self.channel_create_schema = Schema({
'name': All(str, Length(min=2, max=100)),
_o('type'): int,
_o('bitrate'): int,
_o('user_limit'): int,
_o('permission_overwrites'): list,
}, required=True, extra=REMOVE_EXTRA)
self.register()
def __init__(self, server):
self.server = server
self.guild_man = server.guild_man
self.invite_create_schema = Schema({
Required('max_age', default=86400): All(int, Range(min=10, max=86400)),
Required('max_uses', default=0): All(int, Range(min=0, max=50)),
Required('temporary', default=False): bool,
Required('unique', default=True): bool,
}, extra=REMOVE_EXTRA)
self.register(server.app)
def __init__(self, ws, **kwargs):
super().__init__(ws)
self.ws = ws
self.loop = ws.loop
self.options = kwargs['config']
self.server = kwargs['server']
self.state = None
self.session_id = None
self._encoder, self._decoder = get_data_handlers(self.options[1])
# Client's heartbeat interval, chose at random between 40 and 42sec
self.hb_interval = random.randint(HB_MIN_MSEC, HB_MAX_MSEC)
self.wait_task = None
# ratelimiting tasks that clean the request counter
self.ratelimit_tasks = {}
self.request_counter = {}
# some flags for the client etc
self.identified = False
self.dispatch_lock = asyncio.Lock()
# references to objects
self.guild_man = self.server.guild_man
self.presence = self.server.presence
self.relations = self.server.relations
self.settings = self.server.settings
_o = Optional
self.identify_schema = Schema({
'token': str,
'properties': dict,
_o('compress'): bool,
'large_threshold': int,
'shard': list,
}, extra=REMOVE_EXTRA)
self.resume_schema = Schema({
'token': str,
'session_id': str,
'seq': int
}, extra=REMOVE_EXTRA)
self.req_guild_schema = Schema({
'guild_id': str,
'query': str,
'limit': int,
}, extra=REMOVE_EXTRA)
