Python输入验证和边缘案例处理

Question

有没有比我在这个函数中使用的输入验证更好的模式？

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

这里是没有任何验证相同的功能。它更具可读性，它很短而且重要（9 LoC vs 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) 

我希望它能帮助。