Untitled
unknown
plain_text
6 months ago
1.7 kB
3
Indexable
Never
from cassandra.cqlengine import columns from cassandra.cqlengine.models import Model class HosCartDetails(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() 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 from models import HosCartDetails 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(HosCartDetails) @app.route('/hos_cart_details/<event_date>') def get_hos_cart_details(event_date): # Fetch hoscd_event_date details using the model class results = HosCartDetails.objects.filter(hoscd_event_date=event_date) data = [{'hoscd_businessid': r.hoscd_businessid, 'hoscd_key': r.hoscd_key} for r in results] return jsonify(data) if __name__ == '__main__': app.run(debug=True)