Site icon Learning & Doing

App Dev: Developing a Backend Service – Python

admin
developing

“App Dev: Developing a Backend Service – Python”

Pengantar

Google App Engine memungkinkan Anda mengelola sumber daya dari baris perintah, men-debug kode sumber dalam produksi, dan menjalankan backend API. Lab ini berkonsentrasi pada layanan backend, menyatukan layanan Pub/Sub, Natural Language, dan Spanner serta API untuk mengumpulkan dan menganalisis umpan balik dan skor dari aplikasi Kuis online.

Praktikum

Task 1. Prepare the quiz application

Clone source code in Cloud Shell

git clone https://github.com/GoogleCloudPlatform/training-data-analyst

Configure and run the Quiz application

cd ~/training-data-analyst/courses/developingapps/v1.2/python/pubsub-languageapi-spanner/start

. prepare_web_environment.sh

python run_server.py

cd ~/training-data-analyst/courses/developingapps/v1.2/python/pubsub-languageapi-spanner/start

. run_worker.sh

Check out the Quiz application

Task 2. Examine the Quiz application code

Review the Google Cloud application code structure

Review the web application code

Task 3. Work with Cloud Pub/Sub

Create a Cloud Pub/Sub topic

Create a Cloud Pub/Sub subscription

gcloud pubsub subscriptions create worker-subscription --topic feedback

Publish a message to a Cloud Pub/Sub topic

gcloud pubsub topics publish feedback --message "Hello World"

Retrieve a message from a Cloud Pub/Sub subscription

gcloud beta pubsub subscriptions pull worker-subscription --auto-ack

Task 4. Publish messages to Cloud Pub/Sub programmatically

Import and use the Python Cloud Pub/Sub module

Baca Juga :  App Dev: Storing Application Data in Cloud Datastore - Python

quiz/gcp/pubsub.py

# TODO: Load the Cloud Pub/Sub module
from google.cloud import pubsub_v1
# END TODO

# TODO: Create a Pub/Sub Publisher Client
publisher = pubsub_v1.PublisherClient()
# END TODO
# TODO: Create Topic Object to reference feedback topic
topic_path = publisher.topic_path(project_id, 'feedback')
# END TODO

Write code to publish a message to Cloud Pub/Sub

quiz/gcp/pubsub.py

"""
Publishes feedback info
- jsonify feedback object
- encode as bytestring
- publish message
- return result
"""
def publish_feedback(feedback):
# TODO: Publish the feedback object to the feedback topic
    payload = json.dumps(feedback, indent=2,
                                      sort_keys=True)
    data = payload.encode('utf-8')
    future = publisher.publish(topic_path, data=data)
    return future.result()
# END TODO

Write code to use the Pub/Sub publish functionality

quiz/api/api.py

# TODO: Add pubsub to import list
from quiz.gcp import datastore, pubsub
# END TODO

quiz/api/api.py__, publish_feedback(…) function

"""
Publish feedback
- Call pubsub helper
- Compose and return response
"""
def publish_feedback(feedback):
    # TODO: Publish the feedback using your pubsub module,
    # return the result
    result = pubsub.publish_feedback(feedback)
    response = Response(json.dumps(
                        result, indent=2, sort_keys=True))
    response.headers['Content-Type'] = 'application/json'
    return response
    # END TODO

Run the application and create a Pub/Sub message

gcloud pubsub subscriptions pull worker-subscription --auto-ack

Task 5. Subscribe to Cloud Pub/Sub topics programmatically

Write code to create a Cloud Pub/Sub subscription and receive messages

/quiz/gcp/pubsub.py

# TODO: Create a Pub/Sub Subscriber Client
sub_client = pubsub_v1.SubscriberClient()
# END TODO
# TODO: Create a Subscription object named
# worker-subscription
sub_path = sub_client.subscription_path(project_id, 'worker-subscription')
# END TODO
def pull_feedback(callback):
    # TODO: Subscribe to the worker-subscription,
    # invoking the callback
    sub_client.subscribe(sub_path, callback=callback)
    # END TODO

Write code to use the Pub/Sub subscribe functionality

console/worker.py

# TODO: Load the pubsub, languageapi and spanner modules from
# from the quiz.gcp package
from quiz.gcp import pubsub
# END TODO
def pubsub_callback(message):
    # TODO: Acknowledge the message
    message.ack()
    # END TODO
    log.info('Message received')
    # TODO: Log the message
    log.info(message)
    # END TODO
def main():
    log.info('Worker starting...')
    # TODO: Register the callback
    pubsub.pull_feedback(pubsub_callback)
    # END TODO
    while True:
        time.sleep(60)

Run the web and worker application and create a Pub/Sub message

python run_server.py
. run_worker.sh
INFO:root:Worker starting...
INFO:root:Message received
INFO:root:Message {
 data: b'{\n  "email": "app.dev.student@example.org",\n  "fee...'
 ordering_key: ''
 attributes: {}
}
Baca Juga :  Setting up a Private Kubernetes Cluster

Task 6. Use the Cloud Natural Language API

Write code to invoke the Cloud Natural Language API

quiz/gcp/languageapi.py

# TODO: Import the language module
from google.cloud import language_v1
# END TODO
# TODO: Create the Language API client
lang_client = language_v1.LanguageServiceClient()
# END TODO
def analyze(text):
    # TODO: Create a Document object
    doc = language_v1.types.Document(content=text,
                    type_='PLAIN_TEXT')
    # END TODO
    # TODO: Analyze the sentiment
    sentiment = lang_client.analyze_sentiment(
                    document=doc).document_sentiment
    # END TODO
    # TODO: Return the sentiment score
    return sentiment.score
    # END TODO

Write code to use the Natural Language API functionality

console/worker.py

# TODO: Load the pubsub, languageapi and spanner modules from
# from the quiz.gcp package
from quiz.gcp import pubsub, languageapi
# END TODO
def pubsub_callback(message):
    # TODO: Acknowledge the message
    message.ack()
    # END TODO
    log.info('Message received')
    # TODO: Log the message
    log.info(message)
    # END TODO
    data = json.loads(message.data)
    # TODO: Use the languageapi module to
    # analyze the sentiment
    score = languageapi.analyze(str(data['feedback']))
    # END TODO
    # TODO: Log the sentiment score
    log.info('Score: {}'.format(score))
    # END TODO
    # TODO: Assign the sentiment score to
    # a new score property
    data['score'] = score
    # END TODO

Run the web and worker application and test the Natural Language API

Task 7. Persist data to Cloud Spanner

Create a Cloud Spanner instance

Create a Cloud Spanner database and table

CREATE TABLE Feedback (
    feedbackId STRING(100) NOT NULL,
    email STRING(100),
    quiz STRING(20),
    feedback STRING(MAX),
    rating INT64,
    score FLOAT64,
    timestamp INT64
)
PRIMARY KEY (feedbackId);

Write code to persist data into Cloud Spanner

Baca Juga :  Building a VPN Between Google Cloud and AWS with Terraform

quiz/gcp/spanner.py

import re
# TODO: Import the spanner module
from google.cloud import spanner
# END TODO
# TODO: Create a spanner Client
spanner_client = spanner.Client()
# END TODO
# TODO: Get a reference to the Cloud Spanner quiz-instance
instance = spanner_client.instance('quiz-instance')
# END TODO
# TODO: Get a reference to the Cloud Spanner quiz-database
database = instance.database('quiz-database')
# END TODO

quiz/gcp/spanner.py

def save_feedback(data):
    # TODO: Create a batch object for database operations
    with database.batch() as batch:
    # END TODO
        # TODO: Create a key for the record
        # from the email, quiz and timestamp
        feedback_id = '{}_{}_{}'.format(
                    reverse_email(data['email']),
                    data['quiz'],
                    data['timestamp'])
        # END TODO
        # TODO: Use the batch to insert a record
        # into the feedback table
        # This needs the columns and values
        batch.insert(
            table='feedback',
            columns=(
                'feedbackId',
                'email',
                'quiz',
                'timestamp',
                'rating',
                'score',
                'feedback'
            ),
            values=[
                (
                    feedback_id,
                    data['email'],
                    data['quiz'],
                    data['timestamp'],
                    data['rating'],
                    data['score'],
                    data['feedback']
                )
            ]
        )
        # END TODO

Write code to use the Cloud Spanner functionality

quiz/console/worker.py

# TODO: Load the pubsub, languageapi and spanner modules
# from the quiz.gcp package
from quiz.gcp import pubsub, languageapi, spanner
# END TODO
logging.basicConfig(stream=sys.stdout, level=logging.INFO)
log = logging.getLogger()
def pubsub_callback(message):
    # TODO: Acknowledge the message
    message.ack()
    # END TODO
    log.info('Message received')
    # TODO: Log the message
    log.info(message)
    # END TODO
    data = json.loads(message.data)
    # TODO: Use the languageapi module to
    # analyze the sentiment
    score = languageapi.analyze(str(data['feedback']))
    # END TODO
    # TODO: Log the sentiment score
    log.info('Score: {}'.format(score))
    # END TODO
    # TODO: Assign the sentiment score to
    # a new score property
    data['score'] = score
    # END TODO
    # TODO: Use the spanner module to save the feedback
    spanner.save_feedback(data)
    # END TODO
    # TODO: Log a message to say the feedback
    # has been saved
    log.info('Feedback saved')
    # END TODO

Run the web and worker application and test Cloud Spanner

SELECT * FROM Feedback

Penutup

Sahabat Blog Learning & Doing demikianlah penjelasan mengenai App Dev: Developing a Backend Service – Python. Semoga Bermanfaat . Sampai ketemu lagi di postingan berikut nya.

Exit mobile version