Cloudera ऑपरेशनल डेटाबेस और फ्लास्क का उपयोग करके एक साधारण CRUD वेब एप्लिकेशन और इमेज स्टोर का निर्माण

8. अपने ब्राउज़र में http://localhost:8888/users पर जाएं। आपको एप्लिकेशन को चलते हुए देखने में सक्षम होना चाहिए! उतना ही सरल।

संहिता का अध्ययन करना

1. स्कीमा वर्ग, मूल रूप से कनेक्शन विवरण रखता है और तालिका विधियाँ बनाता और छोड़ता है। जैसा कि आप देख सकते हैं, "फोटो" कॉलम एक VARBINARY प्रकार है, जो HBase में एक MOB ऑब्जेक्ट में अनुवाद करता है:

import phoenixdb
import phoenixdb.cursor
class Schema:
    def __init__(self):
        opts = {}
        opts['authentication'] = 'BASIC'
        opts['avatica_user'] = '<cod workload username>'
        opts['avatica_password'] = '<cod workload pw>'
        database_url = "<cod thin jdbc url>"
        self.TABLENAME = "users"
        self.conn = phoenixdb.connect(database_url, autocommit=True,**opts)
        self.curs = self.conn.cursor()

    def create_users_table(self):
        query = """
        CREATE TABLE IF NOT EXISTS """+self.TABLENAME+""" (
        username VARCHAR NOT NULL,
        firstname VARCHAR,
        lastname  VARCHAR,
        telephone VARCHAR,
        message VARCHAR,
        email VARCHAR,
        photo VARBINARY,
        photo_name VARCHAR,
        photo_type VARCHAR,
        photo_chars VARCHAR
        CONSTRAINT my_pk PRIMARY KEY (username))
        """
        self.curs.execute(query)

    def drop_users_table(self):
        query = "DROP TABLE "+self.TABLENAME
        self.curs.execute(query)

2 फीनिक्स के साथ सभी एप्लिकेशन संचालन के लिए उपयोगकर्ता वर्ग जिम्मेदार है। हम अद्यतन/सम्मिलित कर सकते हैं (फीनिक्स भाषा में अप्सर्ट), हटा सकते हैं, सूचीबद्ध कर सकते हैं और छवियों के लेनदेन को संभाल सकते हैं:

import phoenixdb
from schema import Schema
import json
class UsersModel:
    TABLENAME = "users"

    def __init__(self):
        db = Schema()
        self.conn=db.conn
        self.curs=db.curs

    def upsert(self, params):

        sql = "upsert into " + self.TABLENAME + \
            " (username ,message,telephone,firstname,lastname,email) \
             values (?,?,?,?,?,?)"
        data = (params.get('username'),params.get('message'),\
            params.get('telephone'),params.get('firstname'),\
            params.get('lastname'),params.get('email'))
        results = self.curs.execute(sql,data)
        return results

    def upsert_photo(self, params):
        if params.get('photo') is None:
            photo = bytes('','utf-8')
        else:
            photo = params.get('photo')

        sql = "upsert into " + self.TABLENAME + \
            " (username, photo,photo_name) values (?,?,?)"

        data = (params.get('username'),photo, params.get('photo_name'))
        results = self.curs.execute(sql,data)
        return results

    def delete(self, username):
        query = f"DELETE from {self.TABLENAME} " \
                f"WHERE username = {username}"

        self.curs.execute(query)

    def list_items(self, where_clause="",format="json"):
        query = f"SELECT username ,email,message,telephone,firstname,\
            lastname,photo_name " \
            f"from {self.TABLENAME} WHERE  " + where_clause

        self.curs.execute(query)
        if format=="json":
            r = [dict((self.curs.description[i][0].lower(), value) \
                   for i, value in enumerate(row)) for row in \
                   self.curs.fetchall()]
            self.conn.close()
            data={'data': r }
            return json.dumps(data)

        result_set=self.curs.fetchall()
        result = [{column: row[i]
            for i, column in enumerate(result_set[0].keys())}
                for row in result_set]
        return result
    def get_image(self, username):
        query = f"SELECT photo,photo_name " \
                f"from {self.TABLENAME} WHERE  username='"+username+"'"

        self.curs.execute(query)
        row = self.curs.fetchone()
        return row

3. app.py एप्लिकेशन का मुख्य राउटर है। इसमें उपयोगकर्ता इनपुट के साथ सभी हैंडलिंग और उन्हें कनेक्ट विधियों में रूट करना शामिल है। मैंने उपयोग में आसानी के लिए छवियों के संचालन को अलग कर दिया, और इस तरह मैं एक उपयोगकर्ता के लिए एक विशिष्ट छवि प्राप्त कर सकता हूं:

from flask import Flask, request, send_file ,jsonify,render_template
import phoenixdb
import io
from users import UsersModel
from schema import Schema
import json

app = Flask(__name__)

@app.after_request
def add_headers(response):
    response.headers['Access-Control-Allow-Origin'] = '*'
    response.headers['Access-Control-Allow-Headers'] =  \
        "Content-Type, Access-Control-Allow-Headers, Authorization, \
        X-Requested-With"
    response.headers['Access-Control-Allow-Methods']=  "POST, GET, PUT, \
    DELETE, OPTIONS"
    response.headers['Allow']=  "POST, GET, PUT, OPTIONS"
    return response

@app.route("/")
def hello():
    return "Hello World!"

@app.route("/users")
def return_form():
    return render_template("users.html")

@app.route("/handle_data",methods=['POST'])
def handle_data():
    if request.method == 'POST':
        username = request.form['username']
        firstname = request.form['firstname']
        lastname = request.form['lastname']
        email = request.form['email']
        telephone = request.form['telephone']
        message = request.form['message']
        photo = request.files['photo']
        photo_bytes = photo.read()
        model=Schema()
        usersmodel=UsersModel()
        data = {'username':f"{username}",'firstname':f"{firstname}",\
            'lastname':f"{lastname}",'telephone':f"{telephone}",\
            'message':f"{message}"}
        photo_data = {'username':f"{username}",\
            'photo':photo_bytes,\
            'photo_name':f"{photo.filename}"}
        usersmodel.upsert(data)
        usersmodel.upsert_photo(photo_data)
        return render_template('users.html')
    else:
        return render_template('users.html')

@app.route("/get_users",methods=['GET'])
def get_users():
    if request.method == 'GET':
        usersmodel=UsersModel()
        users = usersmodel.list_items("1=1")
        return users

@app.route("/get_image",methods=['GET'])
def get_image():
    if request.method == 'GET':
        username = request.args.get('username')
        usersmodel=UsersModel()
        imagedb = usersmodel.get_image(username)
        return send_file(io.BytesIO(imagedb[0]),mimetype='image/png', \
            attachment_filename=imagedb[1])

if __name__ == "__main__":
    Schema()
    app.run(debug=True, port=8888)

अगले चरण में, आप अपने आवेदन का परीक्षण करने के लिए इस जीथब रेपो का उपयोग कर सकते हैं।

आशा है कि आपको यह उपयोगी लगेगा, हैप्पी कोडिंग !!