Untitled
unknown
plain_text
7 months ago
3.4 kB
3
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 # Define the model class class HosCartDetails(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() # Create the Flask app app = Flask(__name__) # Load the configuration file app.config.from_pyfile('config.cfg') # Create the Cassandra cluster and session cluster = Cluster(app.config['CASSANDRA_CLUSTER'].split(',')) session = cluster.connect() # Set the keyspace session.set_keyspace(app.config['CASSANDRA_KEYSPACE']) # 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 = HosCartDetails.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 = HosCartDetails.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 = HosCartDetails.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 = HosCartDetails.objects.filter(hoscd_username=username).allow_filtering() # Convert the rows to a JSON response response = [] for row in rows: response.append(row.__dict__)