Untitled
unknown
plain_text
7 months ago
2.3 kB
1
Indexable
Never
from flask import Flask, request, jsonify from cassandra.cluster import Cluster from cassandra.auth import PlainTextAuthProvider #from utils.hos_cart_details_models import HosCartDetails from cassandra.cqlengine import columns from cassandra.cqlengine.models import Model # Define the model class class hos_cart_details(Model): __keyspace__='hobs_ods_staging' 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__) app.config['CASSANDRA_HOSTS'] = ['172.6.177.58'] app.config['CASSANDRA_KEYSPACE'] = 'hobs_ods_staging' 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() @app.route('/api/businessid/<businessid>', methods=['GET']) def get_by_businessid(businessid): results = hos_cart_details.objects(hoscd_businessid=businessid).all() return jsonify([result.to_dict() for result in results]) @app.route('/api/username/<username>', methods=['GET']) def get_by_username(username): results = hos_cart_details.objects(hoscd_username=username).all() return jsonify([result.to_dict() for result in results]) @app.route('/api/event_date/<event_date>', methods=['GET']) def get_by_event_date(event_date): results = hos_cart_details.objects(hoscd_event_date=event_date).all() return jsonify([result.to_dict() for result in results]) @app.route('/api/status/<status>', methods=['GET']) def get_by_status(status): results = hos_cart_details.objects(hoscd_status=status).all() return jsonify([result.to_dict() for result in results]) if __name__ == '__main__': app.run(host='172.16.177.58',port=11421)