Untitled
unknown
plain_text
7 months ago
3.5 kB
1
Indexable
Never
from flask import Flask, jsonify from datetime import datetime from cassandra.cluster import Cluster from cassandra.cqlengine import columns from cassandra.cqlengine.management import sync_table from cassandra.cqlengine.models import Model from cassandra.auth import PlainTextAuthProvider # Define the model class class hos_cart_details(Model): hoscd_event_date = columns.Date(primary_key=True, partition_key=True) hoscd_businessid = columns.Text(primary_key=True, partition_key=True) hoscd_key = columns.Text(primary_key=True, clustering_order="ASC") 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__) contact_points = ['172.16.177.58'] auth_provider = PlainTextAuthProvider('hobs_analytics', 'hobs_analytics') cluster = Cluster(contact_points=contact_points, auth_provider=auth_provider) session = cluster.connect() # Sync the HosCartDetails model with the Cassandra table sync_table(HosCartDetails) # Define the API endpoints @app.route('/hos_cart_details/date/<string:date>', methods=['GET']) def get_hos_cart_details_by_date(date): try: # Parse the date parameter event_date = datetime.strptime(date, '%Y-%m-%d').date() # Fetch the data from the Cassandra table rows = hos_cart_details.objects.filter(hoscd_event_date=event_date).allow_filtering() # Convert the rows to a JSON response response = [] for row in rows: response.append(row.__dict__) return jsonify(response) except Exception as e: return jsonify({'error': str(e)}), 500 @app.route('/hos_cart_details/businessid/<string:businessid>', methods=['GET']) def get_hos_cart_details_by_businessid(businessid): try: # Fetch the data from the Cassandra table rows = hos_cart_details.objects.filter(hoscd_businessid=businessid).allow_filtering() # Convert the rows to a JSON response response = [] for row in rows: response.append(row.__dict__) return jsonify(response) except Exception as e: return jsonify({'error': str(e)}), 500 @app.route('/hos_cart_details/status/<string:status>', methods=['GET']) def get_hos_cart_details_by_status(status): try: # Fetch the data from the Cassandra table rows = hos_cart_details.objects.filter(hoscd_status=status).allow_filtering() # Convert the rows to a JSON response response = [] for row in rows: response.append(row.__dict__) return jsonify(response) except Exception as e: return jsonify({'error': str(e)}), 500 @app.route('/hos_cart_details/username/<string:username>', methods=['GET']) def get_hos_cart_details_by_username(username): try: # Fetch the data from the Cassandra table rows = hos_cart_details.objects.filter(hoscd_username=username).allow_filtering() # Convert the rows to a JSON response response = [] for row in rows: response.append(row.__dict__) return jsonify(response) except Exception as e: return jsonify({'error': str(e)}), 500