Versuchen Sie es mit Amazon DynamoDB von Python

0. Einführung

Probieren wir verschiedene von AWS serverless empfohlene Dienste aus. Verwenden wir dieses Mal Amazon DynamoDB von Python.

1. 1. Erstellen Sie eine Python-Umgebung auf Ihrem Mac

Es gibt überhaupt keine Umgebung !! Also, es ist von der Umweltkonstruktion.

Erstellen Sie die Umgebung unter Bezugnahme auf Folgendes (oder wie es ist ...).

Wenn Sie .bash_profile nicht festlegen, müssen Sie anscheinend jedes Mal den Befehl drücken, um die Python-Version zu wechseln. Legen Sie dies also ebenfalls fest.

2. Erstellen Sie ein Skript, das Amazon DynamoDB in der Python-Umgebung von Mac aufruft

Erstellen Sie ein Python-Skript, indem Sie auf Folgendes verweisen.

Da ich von einem lokalen Mac stamme, ist die IAM-Rolle nicht verfügbar, und ich wollte alles mit einem einzigen Skript abschließen, daher verwende ich diese Authentifizierungsmethode.

python


import boto3
from boto3.session import Session

accesskey = "YOUR_ACCESSKEY"
secretkey = "YOUR_SECRETKEY"
region    = "YOUR_REGION"

session = Session(
                  aws_access_key_id=accesskey,
                  aws_secret_access_key=secretkey,
                  region_name=region)

3. 3. Befolgen Sie das Amazon DynamoDB-Handbuch Erste Schritte in der AWS-Dokumentation, um zu erfahren, wie Sie mit DynamoDB arbeiten

Bestätigen Sie den Vorgang wie von AWS angegeben.

  1. Schritt 1: Erstellen Sie eine Tabelle - Amazon DynamoDB
  2. Schritt 2: Laden von Beispieldaten - Amazon DynamoDB
  3. Schritt 3: Erstellen, Laden, Aktualisieren, Löschen von Elementen - Amazon DynamoDB
  4. Schritt 4: Daten abfragen und scannen - Amazon DynamoDB
  5. Schritt 5: (Optional) Drop-Tabelle - Amazon DynamoDB

Unten ist ein Beispiel. Ich habe es bestätigt, indem ich den Aufruf der Hauptfunktion kommentiert habe.

Um ehrlich zu sein, bin ich nicht an Python gewöhnt, daher bin ich mir nicht sicher, ob ich es so schreiben kann ...

SamplaDynamoDB.py


#!/usr/bin/env python
# -*- coding: utf-8 -*-

# ---1----+----2----+----3----+----4----+----5----+----6----+----7----+----8----
# ==============================================================================
#
# SampleDynamoDB.py
#
#   * ex) $ ./SamplaDynamoDB.py
#
# ==============================================================================

import sys
import io
import logging
import json
import decimal
import time

import datetime
from datetime import datetime as dt

import boto3
from boto3.session import Session
from boto3.dynamodb.conditions import Key, Attr

# Helper class to convert a DynamoDB item to JSON.
class DecimalEncoder(json.JSONEncoder):
    def default(self, o):
        if isinstance(o, decimal.Decimal):
            if o % 1 > 0:
                return float(o)
            else:
                return int(o)
        return super(DecimalEncoder, self).default(o)

sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding='utf-8')
logging.basicConfig(level=logging.INFO, filename=(__file__ + ".log"), format="%(asctime)s %(levelname)s %(filename)s %(lineno)d %(funcName)s | %(message)s")

# ------------------------------------------------------------------------------
# Set
# ------------------------------------------------------------------------------
tmp_today = datetime.datetime.today()

# AWS
accesskey = "[Zugriffsschlüssel-ID]"
secretkey = "[Geheimer Zugangsschlüssel]"
region = "ap-northeast-1"
session = Session(aws_access_key_id=accesskey, aws_secret_access_key=secretkey, region_name=region)
dynamodb = session.resource('dynamodb')

# ------------------------------------------------------------------------------
#"Schritt 1:Erstellen Sie eine Tabelle- Amazon DynamoDB」
# <http://docs.aws.amazon.com/ja_jp/amazondynamodb/latest/gettingstartedguide/GettingStarted.Python.01.html>
# ------------------------------------------------------------------------------
def MoviesCreateTable():
    logging.info("<<<<<<<< %s Start >>>>>>>>", __name__)
    try:
        table = dynamodb.create_table(
            TableName='Movies',
            KeySchema=[
                {
                    'AttributeName': 'year',
                    'KeyType': 'HASH'  #Partition key
                },
                {
                    'AttributeName': 'title',
                    'KeyType': 'RANGE'  #Sort key
                }
            ],
            AttributeDefinitions=[
                {
                    'AttributeName': 'year',
                    'AttributeType': 'N'
                },
                {
                    'AttributeName': 'title',
                    'AttributeType': 'S'
                },
            ],
            ProvisionedThroughput={
                'ReadCapacityUnits': 10,
                'WriteCapacityUnits': 10
            }
        )
        logging.info("Table status : [%s]", table.table_status)
    except Exception as e:
        logging.error("Type : %s", type(e))
        logging.error(e)
    logging.info("<<<<<<<< %s End >>>>>>>>", __name__)

# ------------------------------------------------------------------------------
#"Schritt 2:Beispieldaten laden- Amazon DynamoDB」
# <http://docs.aws.amazon.com/ja_jp/amazondynamodb/latest/gettingstartedguide/GettingStarted.Python.02.html>
# ------------------------------------------------------------------------------
def MoviesLoadData():
    logging.info("<<<<<<<< %s Start >>>>>>>>", __name__)
    try:
        with open("moviedata.json") as json_file:
            movies = json.load(json_file, parse_float = decimal.Decimal)
            for movie in movies:
                year = int(movie['year'])
                title = movie['title']
                info = movie['info']
                #logging.info("Adding Movie | Year:[%s], Title:[%s]", year, title)
                table.put_item(
                   Item={
                       'year': year,
                       'title': title,
                       'info': info,
                    }
                )
    except Exception as e:
        logging.error("Type : %s", type(e))
        logging.error(e)
    logging.info("<<<<<<<< %s End >>>>>>>>", __name__)

# ------------------------------------------------------------------------------
#"Schritt 3:Elemente erstellen, laden, aktualisieren, löschen- Amazon DynamoDB」
# <http://docs.aws.amazon.com/ja_jp/amazondynamodb/latest/gettingstartedguide/GettingStarted.Python.03.html>
# ------------------------------------------------------------------------------
def MoviesItemOps01():
    logging.info("<<<<<<<< %s Start >>>>>>>>", __name__)
    try:
        table = dynamodb.Table('Movies')
        title = "The Big New Movie"
        year = 2015
        response = table.put_item(
           Item={
                'year': year,
                'title': title,
                'info': {
                    'plot':"Nothing happens at all.",
                    'rating': decimal.Decimal(0)
                }
            }
        )
        logging.info("PutItem succeeded:")
        logging.info(json.dumps(response, indent=4, cls=DecimalEncoder))
    except Exception as e:
        logging.error("type : %s", type(e))
        logging.error(e)
    logging.info("<<<<<<<< %s End >>>>>>>>", __name__)

def MoviesItemOps02():
    logging.info("<<<<<<<< %s Start >>>>>>>>", __name__)
    try:
        table = dynamodb.Table('Movies')
        title = "The Big New Movie"
        year = 2015
        try:
            response = table.get_item(
                Key={
                    'year': year,
                    'title': title
                }
            )
        except ClientError as e:
            logging.info(e.response['Error']['Message'])
        else:
            item = response['Item']
            logging.info("GetItem succeeded:")
            logging.info(json.dumps(item, indent=4, cls=DecimalEncoder))
    except Exception as e:
        logging.error("type : %s", type(e))
        logging.error(e)
    logging.info("<<<<<<<< %s End >>>>>>>>", __name__)

def MoviesItemOps03():
    logging.info("<<<<<<<< %s Start >>>>>>>>", __name__)
    try:
        table = dynamodb.Table('Movies')
        title = "The Big New Movie"
        year = 2015
        response = table.update_item(
            Key={
                'year': year,
                'title': title
            },
            UpdateExpression="set info.rating = :r, info.plot=:p, info.actors=:a",
            ExpressionAttributeValues={
                ':r': decimal.Decimal(5.5),
                ':p': "Everything happens all at once.",
                ':a': ["Larry", "Moe", "Curly"]
            },
            ReturnValues="UPDATED_NEW"
        )
        logging.info("UpdateItem succeeded:")
        logging.info(json.dumps(response, indent=4, cls=DecimalEncoder))
    except Exception as e:
        logging.error("type : %s", type(e))
        logging.error(e)
    logging.info("<<<<<<<< %s End >>>>>>>>", __name__)

def MoviesItemOps04():
    logging.info("<<<<<<<< %s Start >>>>>>>>", __name__)
    try:
        table = dynamodb.Table('Movies')
        title = "The Big New Movie"
        year = 2015
        response = table.update_item(
            Key={
                'year': year,
                'title': title
            },
            UpdateExpression="set info.rating = info.rating + :val",
            ExpressionAttributeValues={
                ':val': decimal.Decimal(1)
            },
            ReturnValues="UPDATED_NEW"
        )
        logging.info("UpdateItem succeeded:")
        logging.info(json.dumps(response, indent=4, cls=DecimalEncoder))
    except Exception as e:
        logging.error("type : %s", type(e))
        logging.error(e)
    logging.info("<<<<<<<< %s End >>>>>>>>", __name__)

def MoviesItemOps05():
    logging.info("<<<<<<<< %s Start >>>>>>>>", __name__)
    try:
        table = dynamodb.Table('Movies')
        title = "The Big New Movie"
        year = 2015
        logging.info("Attempting conditional update...")
        try:
            response = table.update_item(
                Key={
                    'year': year,
                    'title': title
                },
                UpdateExpression="remove info.actors[0]",
                ConditionExpression="size(info.actors) > :num",
                ExpressionAttributeValues={
                    ':num': 2
                },
                ReturnValues="UPDATED_NEW"
            )
        except ClientError as e:
            if e.response['Error']['Code'] == "ConditionalCheckFailedException":
                logging.error(e.response['Error']['Message'])
            else:
                raise
        else:
            logging.info("UpdateItem succeeded:")
            logging.info(json.dumps(response, indent=4, cls=DecimalEncoder))
    except Exception as e:
        logging.error("type : %s", type(e))
        logging.error(e)
    logging.info("<<<<<<<< %s End >>>>>>>>", __name__)

def MoviesItemOps06():
    logging.info("<<<<<<<< %s Start >>>>>>>>", __name__)
    try:
        table = dynamodb.Table('Movies')
        title = "The Big New Movie"
        year = 2015
        logging.info("Attempting a conditional delete...")
        try:
            response = table.delete_item(
                Key={
                    'year': year,
                    'title': title
                },
                ConditionExpression="info.rating <= :val",
                ExpressionAttributeValues= {
                    ":val": decimal.Decimal(8)
                }
            )
        except ClientError as e:
            if e.response['Error']['Code'] == "ConditionalCheckFailedException":
                logging.info(e.response['Error']['Message'])
            else:
                raise
        else:
            logging.info("DeleteItem succeeded:")
            logging.info(json.dumps(response, indent=4, cls=DecimalEncoder))
    except Exception as e:
        logging.error("type : %s", type(e))
        logging.error(e)
    logging.info("<<<<<<<< %s End >>>>>>>>", __name__)

# ------------------------------------------------------------------------------
#"Schritt 4:Daten abfragen und scannen- Amazon DynamoDB」
# <http://docs.aws.amazon.com/ja_jp/amazondynamodb/latest/gettingstartedguide/GettingStarted.Python.04.html>
# ------------------------------------------------------------------------------
def MoviesQuery01():
    logging.info("<<<<<<<< %s Start >>>>>>>>", __name__)
    try:
        table = dynamodb.Table('Movies')
        logging.info("Movies from 1933")
        response = table.query(
            KeyConditionExpression=Key('year').eq(1933)
        )
        logging.info("Query01 succeeded:")
        logging.info(json.dumps(response, indent=4, cls=DecimalEncoder))
        for i in response['Items']:
            logging.info("%s : %s", i['year'], i['title'])
    except Exception as e:
        logging.error("Type : %s", type(e))
        logging.error(e)
    logging.info("<<<<<<<< %s End >>>>>>>>", __name__)

def MoviesQuery02():
    logging.info("<<<<<<<< %s Start >>>>>>>>", __name__)
    try:
        table = dynamodb.Table('Movies')
        logging.info("Movies from 1992 - titles A-L, with genres and lead actor")
        response = table.query(
            ProjectionExpression="#yr, title, info.genres, info.actors[0]",
            ExpressionAttributeNames={ "#yr": "year" }, # Expression Attribute Names for Projection Expression only.
            KeyConditionExpression=Key('year').eq(1992) & Key('title').between('A', 'L')
        )
        logging.info("Query02 succeeded:")
        logging.info(json.dumps(response, indent=4, cls=DecimalEncoder))
        for i in response[u'Items']:
            logging.info(json.dumps(i, cls=DecimalEncoder))
    except Exception as e:
        logging.error("Type : %s", type(e))
        logging.error(e)
    logging.info("<<<<<<<< %s End >>>>>>>>", __name__)

# ------------------------------------------------------------------------------
#"Schritt 5: (Möglichkeit)Lass den Tisch fallen- Amazon DynamoDB」
# <http://docs.aws.amazon.com/ja_jp/amazondynamodb/latest/gettingstartedguide/GettingStarted.Python.05.html>
# ------------------------------------------------------------------------------
def MoviesDeleteTable():
    logging.info("<<<<<<<< %s Start >>>>>>>>", __name__)
    try:
        table = dynamodb.Table('Movies')
        table.delete()
    except Exception as e:
        logging.error("Type : %s", type(e))
        logging.error(e)
    logging.info("<<<<<<<< %s End >>>>>>>>", __name__)

# ------------------------------------------------------------------------------
# Main
# ------------------------------------------------------------------------------
if __name__ == '__main__':
    logging.info("<<<<<<<< %s Start >>>>>>>>", __name__)

    # Set
    logging.info("REGION : [%s]", region)

    # Args
    logging.info("Argc : [%d]", len(sys.argv))
    for i in range(len(sys.argv)):
        logging.info("Argv[%d] : [%s]", i, sys.argv[i])

    # Check Table
    table = dynamodb.Table('Movies')
    logging.info("Table :")
    logging.info(table)

    # Create Table
    #MoviesCreateTable()
    #time.sleep(9)

    # Load Json Data
    #MoviesLoadData()

    # Query Data
    MoviesItemOps01()
    MoviesItemOps02()
    MoviesItemOps03()
    MoviesItemOps04()
    #MoviesItemOps05()
    #MoviesItemOps06()

    # Query Data
    MoviesQuery01()
    MoviesQuery02()

    # Delete Table
    #MoviesDeleteTable()

    logging.info("<<<<<<<< %s End >>>>>>>>", __name__)

# ---1----+----2----+----3----+----4----+----5----+----6----+----7----+----8----

99. Suchtpunkt

XX. Zusammenfassung

Dieses Mal war ich dankbar, dass alle einführenden Anleitungen die Quelle hatten.

Als nächstes dachte ich, ich würde versuchen, Amazon Lambda zu starten.

Recommended Posts

Versuchen Sie es mit Amazon DynamoDB von Python
Versuchen Sie es mit Tweepy [Python2.7]
[Python] Versuchen Sie, Tkinters Leinwand zu verwenden
Abfrage von Python an Amazon Athena (unter Verwendung des benannten Profils)
Versuchen Sie die IAM-Datenbankauthentifizierung von Python aus
Verwenden von Rstan aus Python mit PypeR
Betreiben Sie DynamoDB von Python aus wie SQL.
Versuchen Sie Python
Versuchen Sie es mit LevelDB mit Python (plyvel)
Verwenden von Cloud-Speicher aus Python3 (Einführung)
Versuchen Sie es mit der Aktions-API von Python argparse
Versuchen Sie es mit dem Python Cmd-Modul
Führen Sie Ansible über Python mithilfe der API aus
Vorsichtsmaßnahmen bei der Verwendung von Phantomjs aus Python
Greifen Sie mit OAuth 2.0 von Python aus auf Tabellenkalkulationen zu
Versuchen Sie es mit LeapMotion mit Python
Versuchen Sie es mit der Wunderlist-API in Python
Von Python bis zur Verwendung von MeCab (und CaboCha)
Versuchen Sie eine Formel mit Σ mit Python
Versuchen Sie, die Kraken-API mit Python zu verwenden
Dialogflow (früher: API.AI) Verwenden Sie das Python SDK #dialogflow
Versuchen Sie, Python mit Google Cloud-Funktionen zu verwenden
Versuchen Sie es mit Junos 'On-Box Python # 2 Commit Script
Ich habe versucht, die UnityCloudBuild-API von Python zu verwenden
Versuchen Sie, Excel mit Python (Xlwings) zu betreiben.
Versuchen Sie, Python von Ruby aus mit Sparsamkeit aufzurufen
Versuchen Sie es mit Junos 'On-Box Python # 1 Op Script
[Amazon Linux] Wechsel von der Python 2-Serie zur Python 3-Serie
Versuchen Sie es mit Tkinter
Versuchen Sie es mit Docker-Py
Versuchen Sie es mit PDFMiner
Starten Sie Python
Versuchen Sie es mit Geopandas
Versuchen Sie es mit Selen
Versuchen Sie es mit scipy
Python> try: / außer:
SQL zu SQL
Versuchen Sie es mit Pythons Webframework Django (1) - Von der Installation bis zum Serverstart
MeCab von Python
Versuchen Sie es mit Django-Swiftbrowser
Versuchen Sie es mit matplotlib
Versuchen Sie es mit tf.metrics
Versuchen Sie es mit PyODE
Scraping mit Python
Versuchen Sie, Studio Library von Python aus zu betreiben. [Anim speichern]
Erstellen eines Nummerierungsprozesses mit Python im lokalen DynamoDB-Nummerierungsprozess
Ich möchte mit Python eine E-Mail von Google Mail senden.
Versuchen Sie es mit der BitFlyer Ligntning API in Python
Versuchen Sie es mit dem Python-Webframework Tornado Part 1
Erstellen Sie mit python3 eine WAV-Datei aus dem GLSL-Shader
Versuchen Sie es mit dem Sammlungsmodul (ChainMap) von python3
Versuchen Sie es mit Tensorflow. ① Erstellen Sie eine Python-Umgebung und führen Sie Tensorflow ein
Versuchen Sie es mit dem Python-Webframework Tornado Part 2
Versuchen Sie, direkt von Python 3 aus auf die YQL-API zuzugreifen
Führen Sie Python-Skripte in Excel aus (mit xlwings).
Versuchen Sie, die ChatWork-API und die Qiita-API in Python zu verwenden
Versuchen Sie, die DropBox Core-API mit Python zu verwenden
Flächenextraktionsmethode mit dem Zellautomaten Versuchen Sie die Flächenextraktion aus dem Bild mit Growcut (Python).
Versuchen Sie, Amazon Timestream auszuführen
Verwenden Sie thingspeak aus Python