Untitled
unknown
plain_text
2 years ago
3.1 kB
7
Indexable
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) == 3: if params[0] == 'eventdate': eventdate = params[1] results = hos_cart_details.objects.filter(hoscd_event_date=eventdate, hoscd_businessid=params[2]).allow_filtering() elif params[0] == 'businessid': businessid = params[1] results = hos_cart_details.objects.filter(hoscd_businessid=businessid, hoscd_event_date=params[2]).allow_filtering() elif params[0] == 'username': username = params[1] results = hos_cart_details.objects.filter(hoscd_username=username, hoscd_businessid=params[2]).allow_filtering() else: return jsonify({'error': 'Invalid path parameters'}) elif 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() elif params[0] == 'username': username = params[1] results = hos_cart_details.objects.filter(hoscd_username=username).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': base
Editor is loading...