Python 输入验证和边缘情况处理

Question

是否有比我在此函数中使用的更好的输入验证模式？

https://github.com/nathancahill/clearbit-intercom/blob/133e4df0cfd1a146cedb3c749fc1b4fac85a6e1b/server.py#L71

这是没有任何验证的相同函数。它更具可读性，简短且切中要点(9 LoC 与 53 LoC)。

def webhook(clearbitkey, appid, intercomkey):
    event = request.get_json()

    id = event['data']['item']['id']
    email = event['data']['item']['email']

    person = requests.get(CLEARBIT_USER_ENDPOINT.format(email=email), auth=(clearbitkey, '')).json()
    domain = person['employment']['domain']

    company = requests.get(CLEARBIT_COMPANY_ENDPOINT.format(domain=domain), auth=(clearbitkey, '')).json()
    note = create_note(person, company)

    res = requests.post(INTERCOM_ENDPOINT,
                        json=dict(user=dict(id=id), body=note),
                        headers=dict(accept='application/json'),
                        auth=(appid, intercomkey))

    return jsonify(note=res.json())

但是，它不处理任何以下错误:

• 字典 KeyError(尤其是嵌套字典)
• HTTP 错误
• 无效的 JSON
• 意外响应

有更好的模式可以遵循吗？我研究了使用像 voluptous 这样的数据验证库但似乎我仍然遇到同样的冗长问题。

Answer 1

你在 github 上的原始代码对我来说似乎很好。它有点过于复杂，但也可以处理所有错误情况。你可以尝试通过抽象的东西来提高可读性。

只是为了演示，我可能编写如下代码:

class ValidationError(Exception):
    "Raises when data validation fails"
    pass


class CallExternalApiError(Exception):
    "Raises when calling external api fails"
    pass


def get_user_from_event(event):
    """Get user profile from event

    :param dict event: request.get_json() result
    :returns: A dict of user profile
    """
    try:
        event_type = event['data']['item']['type']
    except KeyError:
        raise ValidationError('Unexpected JSON format.')
    if event_type != 'user':
        return ValidationError('Event type is not supported.')

    try:
        id = event['data']['item']['id']
        email = event['data']['item']['email']
    except KeyError:
        return ValidationError('User object missing fields.')
    return {'id': id, 'email': email}


def call_json_api(request_function, api_name, *args, **kwargs):
    """An simple wrapper for sending request

    :param request_function: function used for sending request
    :param str api_name: name for this api call
    """
    try:
        res = request_function(*args, **kwargs)
    except:
        raise CallExternalApiError('API call failed to %s.' % api_name)

    try:
        return res.json()
    except:
        raise CallExternalApiError('Invalid response from %s.' % api_name)


@app.route('/<clearbitkey>+<appid>:<intercomkey>', methods=['POST'])
def webhook(clearbitkey, appid, intercomkey):
    """
    Webhook endpoint for Intercom.io events. Uses this format for Clearbit and
    Intercom.io keys:
    /<clearbitkey>+<appid>:<intercomkey>
    :clearbitkey: Clearbit API key.
    :appid: Intercom.io app id.
    :intercomkey: Intercom.io API key.
    Supports User events, specifically designed for the User Created event.
    Adds a note to the user with their employment and company metrics.
    """
    event = request.get_json()
    try:
        return handle_event(event, clearbitkey, appid, intercomkey)
    except (ValidationError, CallExternalApiError) as e:
        # TODO: include **res_objs in response
        return jsonify(error=str(e))


def handle_event(event):
    """Handle the incoming event
    """
    user = get_user_from_event(event)
    res_objs = dict(event=event)

    person = call_json_api(
        requests.get,
        'Clearbit',
        CLEARBIT_USER_ENDPOINT.format(email=user['email']),
        auth=(clearbitkey, '')
    )

    res_objs['person'] = person
    if 'error' in person:
        raise CallExternalApiError('Error response from Clearbit.')

    domain = person['employment']['domain']
    company = None

    if domain:
        try:
            company = call_json_api(
                requests.get,
                'Clearbit',
                CLEARBIT_COMPANY_ENDPOINT.format(domain=domain),
                auth=(clearbitkey, ''))
            )
            if 'error' in company:
                company = None
        except:
            company = None

    res_objs['company'] = company

    try:
        note = create_note(person, company)
    except:
        return jsonify(error='Failed to generate note for user.', **res_objs)

    result = call_json_api(
        requests.post,
        'Intercom',
        (INTERCOM_ENDPOINT, json=dict(user=dict(id=id), body=note),
        headers=dict(accept='application/json'),
        auth=(appid, intercomkey)
    )
    return jsonify(note=result, **res_objs)

希望对您有帮助。