def order_pubsub()

in cqrs/services/orderinfo/main.py [0:0]

• 76 lines of code
• 11 McCabe index (conditional complexity)

def order_pubsub():
    envelope = request.get_json()
    if not isinstance(envelope, dict) or 'message' not in envelope:
        return error500()
    message = envelope['message']
    if 'data' not in message or 'attributes' not in message:
        return error500()
    attributes = message['attributes']
    if 'event_id' not in attributes or 'event_type' not in attributes:
        return error500()

    order_event = json.loads(base64.b64decode(message['data']).decode('utf-8'))
    event_id = attributes['event_id']
    event_type = attributes['event_type']

    query = ds_client.query(kind='ProcessedEvent')
    query.keys_only()
    query.add_filter('event_id', '=', event_id)
    if list(query.fetch()): # duplicate
        print('Duplicate event {}'.format(event_id))
        return 'Finished.', 200

    if event_type != 'order_create':
        print('Unknown event type {}.format(event_type)')
        return 'Finished.', 200

    customer_id = order_event['customer_id']
    order_id = order_event['order_id']
    customer_id = order_event['customer_id']
    product_id = order_event['product_id']
    number = order_event['number']
    order_date = order_event['order_date']

    # get product_name and unit_price from product service.
    url = PRODUCT_SERVICE_URL + '/api/v1/product/get'
    credentials, _ = google.auth.default()
    auth_req = google.auth.transport.requests.Request()
    id_token = google.oauth2.id_token.fetch_id_token(auth_req, url)
    data = {'product_id': product_id}
    headers = {
        'Content-Type': 'application/json',
        'Authorization': 'Bearer {}'.format(id_token)
    }
    req = urllib.request.Request(url, json.dumps(data).encode(), headers)
    result = None
    try:
        with urllib.request.urlopen(req) as res:
            result = json.load(res)
    except urllib.error.HTTPError as err:
        return 'Finished.', 200

    product_name = result['product_name']
    unit_price = result['unit_price']
    
    order_info = {
        'customer_id': customer_id,
        'order_id': order_id,
        'product_id': product_id,
        'number': number,
        'product_name': product_name,
        'unit_price': unit_price,
        'total_price': unit_price * number,
        'order_date': order_date
    }

    # Insert into bq table
    table_ref = bq_client.dataset('cqrs_example').table('order_information')
    table = bq_client.get_table(table_ref)
    order_info_bq = copy.deepcopy(order_info)
    order_info_bq['order_date'] = datetime.datetime.strptime(
        order_info_bq['order_date'], '%Y-%m-%d').date()
    rows = [order_info_bq]
    bq_client.insert_rows(table, rows)
    
    with ds_client.transaction():
        incomplete_key = ds_client.key('OrderInformationCQRS')
        order_info_entity = datastore.Entity(key=incomplete_key)
        order_info_entity.update(order_info)
        ds_client.put(order_info_entity)

        incomplete_key = ds_client.key('ProcessedEvent')
        entity = datastore.Entity(key=incomplete_key)
        entity.update({
            'event_id': event_id,
            'timestamp': datetime.datetime.utcnow()
        })
        ds_client.put(entity)

    return 'Finished.', 200