cqrs/services/orderinfo/main.py

# Copyright 2020 Google Inc.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.


import base64
import copy
import datetime
import json
import os
import urllib

import google.auth.transport.requests
import google.oauth2.id_token
from google.cloud import datastore, bigquery

from flask import Flask, request


PRODUCT_SERVICE_URL = os.getenv('PRODUCT_SERVICE_URL')


app = Flask(__name__)
ds_client = datastore.Client()
bq_client = bigquery.Client()


def error500():
    resp = {
        'message': 'Internal error occured.'
    }
    return resp, 500


@app.route('/')
def index():
    return 'Order information service for CQRS pattern. '


@app.route('/api/v1/orderinfo/list', methods=['POST'])
@app.route('/orderinfo-service-cqrs/api/v1/orderinfo/list', methods=['POST'])
def order_list():
    json_data = request.get_json()
    customer_id, order_date = None, None
    for key in json_data.keys():
        if key == 'customer_id':
            customer_id = json_data[key]
        elif key == 'order_date':
            order_date = json_data[key]
        else:
            invalid_fields.append(key)
    if customer_id is None or order_date is None:
        return error500()

    query = ds_client.query(kind='OrderInformationCQRS')
    query.add_filter('customer_id', '=', customer_id)
    query.add_filter('order_date', '>=', order_date)
    query.add_filter('order_date', '<', order_date + u'\ufffd')
    orders = []
    for result in query.fetch():
        orders.append({
            'order_id': result['order_id'],
            'customer_id': result['customer_id'],
            'product_id': result['product_id'],
            'product_name': result['product_name'],
            'number': result['number'],
            'unit_price': result['unit_price'],
            'total_price': result['total_price'],
            'order_date': result['order_date'],
        })
    resp = {"order_date": order_date, "orders": orders}

    return resp, 200


@app.route('/api/v1/orderinfo/get', methods=['POST'])
@app.route('/orderinfo-service-cqrs/api/v1/orderinfo/get', methods=['POST'])
def order_get():
    json_data = request.get_json()
    customer_id, order_id = None, None
    invalid_fields = []
    for key in json_data.keys():
        if key == 'customer_id':
            customer_id = json_data[key]
        elif key == 'order_id':
            order_id = json_data[key]
        else:
            invalid_fields.append(key)
    if customer_id is None or order_id is None:
        return error500()

    query = ds_client.query(kind='OrderInformationCQRS')
    query.add_filter('customer_id', '=', customer_id)
    query.add_filter('order_id', '=', order_id)
    resp = None
    for result in query.fetch(): # This should return a single entity.
        resp = {
            'order_id': result['order_id'],
            'customer_id': result['customer_id'],
            'product_id': result['product_id'],
            'product_name': result['product_name'],
            'number': result['number'],
            'unit_price': result['unit_price'],
            'total_price': result['total_price'],
            'order_date': result['order_date'],
        }
        break
    if resp is None:
        resp = {'message': 'The order does not exist.'}

    return resp, 200


# Event handler
@app.route('/api/v1/orderinfo/pubsub', methods=['POST'])
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


if __name__ == "__main__":
    app.run(debug=True, host='0.0.0.0', port=int(os.environ.get('PORT', 8080)))