Python voluptuous 模块，REMOVE_EXTRA 实例源码

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

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)