Untitled

mail@pastecode.io avatar
unknown
plain_text
7 months ago
3.1 kB
1
Indexable
Never
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) == 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()
        else:
            return jsonify({'error': 'Invalid path parameters'})
    elif len(params) == 1:
        if params[0] == 'eventdate':
            results = hos_cart_details.objects.filter(hoscd_event_date=eventdate).allow_filtering()
        else:
            return jsonify({'error': 'Invalid path parameters'})
    elif len(params) ==1:
        if params[0] == 'businessid':
            results = hos_cart_details.objects.filter(hoscd_businessid=businessid).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': base64.b64encode(r.hoscd_value).decode('utf-8'),
        'hoscd_valuetype': r.hoscd_valuetype
    } for r in results]

    count = results.count()

    return jsonify({'data': data, 'count': count})

if __name__ == '__main__':
    app.run(host='172.16.177.58',port=11423)