Untitled
unknown
plain_text
7 months ago
3.1 kB
1
Indexable
Never
from cassandra.cqlengine import columns from cassandra.cqlengine.models import Model from flask import Flask, jsonify from cassandra.cluster import Cluster from cassandra.auth import PlainTextAuthProvider from cassandra.cqlengine import connection from cassandra.cqlengine.management import sync_table import base64 class hos_cart_details(Model): __keyspace__ = 'hobs_ods_staging' hoscd_event_date = columns.Date(primary_key=True) hoscd_businessid = columns.Text(primary_key=True) hoscd_key = columns.Text(primary_key=True) hoscd_buid = columns.Text() hoscd_createdtime = columns.DateTime() hoscd_opid = columns.Text() hoscd_status = columns.Text() hoscd_type = columns.Text() hoscd_updatedtime = columns.DateTime() hoscd_username = columns.Text() hoscd_value = columns.Blob() hoscd_valuetype = columns.Text() app = Flask(__name__) # Connect to the Cassandra cluster auth_provider = PlainTextAuthProvider(username='hobs_analytics', password='hobs_analytics') cluster = Cluster(['172.16.177.58'], auth_provider=auth_provider) session = cluster.connect() # Set up the connection for the model connection.set_session(session) # Sync the table with the model sync_table(hos_cart_details) @app.route('/hos_cart_details/<path:path>') def get_hos_cart_details(path): params = path.split('/') if len(params) == 2: if params[0] == 'eventdate': eventdate = params[1] results = hos_cart_details.objects.filter(hoscd_event_date=eventdate).allow_filtering() elif params[0] == 'businessid': businessid = params[1] results = hos_cart_details.objects.filter(hoscd_businessid=businessid).allow_filtering() else: return jsonify({'error': 'Invalid path parameters'}) elif len(params) == 1: if params[0] == 'eventdate': results = hos_cart_details.objects.filter(hoscd_event_date=eventdate).allow_filtering() else: return jsonify({'error': 'Invalid path parameters'}) elif len(params) ==1: if params[0] == 'businessid': results = hos_cart_details.objects.filter(hoscd_businessid=businessid).allow_filtering() else: return jsonify({'error': 'Invalid path parameters'}) else: return jsonify({'error': 'Invalid path parameters'}) data = [{ 'hoscd_event_date': str(r.hoscd_event_date), 'hoscd_buid': r.hoscd_buid, 'hoscd_createdtime': r.hoscd_createdtime, 'hoscd_opid': r.hoscd_opid, 'hoscd_status': r.hoscd_status, 'hoscd_type': r.hoscd_type, 'hoscd_updatedtime': r.hoscd_updatedtime, 'hoscd_username': r.hoscd_username, 'hoscd_value': base64.b64encode(r.hoscd_value).decode('utf-8'), 'hoscd_valuetype': r.hoscd_valuetype } for r in results] count = results.count() return jsonify({'data': data, 'count': count}) if __name__ == '__main__': app.run(host='172.16.177.58',port=11423)