[AWS] Une histoire qui peut être utile pour ceux qui découvrent Lambda-Python et DynamoDB
introduction
Récemment, je m'intéresse à l'architecture sans serveur. Les 2 entrées suivantes sont des articles sur l'utilisation de Java Script avec Google Apps Script, mais cette fois, je souhaite utiliser Lambda sur AWS pour exécuter du code Python sans serveur. Le contenu de l'article est Python presque ravi de vous rencontrer && AWS J'écrirai un enregistrement jusqu'à ce que les gens qui sont presque ravis de vous rencontrer puissent faire diverses choses avec Lambda et DynamoDB. De plus, au début, j'étais désespéré parce que je ne comprenais pas trop AWS, Lambda a eu une erreur et m'a tué pour la première fois, et je suis intéressé par Python mais je l'ai moins utilisé, alors j'aimerais écrire un article d'un point de vue plutôt débutant. C'est comme écrire beaucoup d'images et de contenus que les débutants peuvent comprendre un par un, donc je pense que les utilisateurs avancés de Pythonist et AWS sont peut-être nouveaux, mais merci.
L'article que j'ai écrit la dernière fois
supposition
AWS Ravi de vous rencontrer Histoire de Python environ 2 semaines Service AWS enregistré, niveau gratuit
AWS Dans mon cas, j'étais confus par le nombre de services AWS, alors j'ai d'abord appris un aperçu des services connectés à quels services. Je pense que ce qui suit sera utile. J'ai également recherché des livres gratuits sur Kindle. "Mais j'ai essayé diverses choses en examinant cet article tout en l'utilisant moi-même.
Résumons "AWS est quoi" en 3 lignes Ce que j'ai fait et ce à quoi j'étais accro pendant la période gratuite de la première année d'AWS
enregistrement
Veuillez vous reporter à un autre article pour l'enregistrement AWS car il est supposé être enregistré.
Création de rôle
C'est beaucoup de problèmes sans un rôle AWS, donc Cette fois, nous allons créer un rôle ayant un accès complet à S3, DynamoDB et Lambda.
IAM de AWS Service List
Rôle >> Créer un nouveau rôle
Nom de rôle Cette fois [lambda_dynamo]
Amazon Lambda >> Sélectionner
Pour le moment, j'ajouterai trois accès complet cette fois. AWSLambdaFullAccess AmazonS3FullAccess AmazonDynamoDBFullAccess
L'étape suivante
Rôle >> lambda_dynamo OK si l'élément appelé est ajouté.
Lambda
C'est un service qui peut exécuter des scripts sans serveur. Pratique.
Structure de frais
En gros, vous pouvez l'utiliser pour tester autant que vous le souhaitez
Châssis sans lambda
| 1 mois | demande | Temps de calcul[GB*Secondes] |
|---|---|---|
| 1,000,000 | 400,000 |
Demandes: nombre total de demandes pour l'ensemble de la fonction Temps de calcul: nombre de mémoire et temps
La mémoire minimale de Lambda étant de 128 Mo, si vous utilisez 3 200 000 secondes () par mois, elle atteindra 400 000 [Go * sec].
(128.0[MB] / 1024[MB/GB]) * 3200000[sec] = 400,000[GB*sec]
Il est donc normal d'exécuter le script pendant environ 888 heures. Utilisons-le régulièrement.
(Veuillez essayer de calculer le temps vous-même!)
Essayez d'utiliser le code de test (erreur Python)
Lambda de AWS Service List
Création de la première fonction
Création d'une fonction Lambda >> lambda-canary
[Paramètres de déclenchement] >> Supprimer >> Suivant Ici, vous pouvez exécuter le script une fois toutes les quelques minutes. Peut être défini plus tard, supprimez-le cette fois.
[Paramètres de fonction] J'ai changé le nom en [sample] pour le moment.
Créez une fonction en faisant du rôle le rôle que vous avez créé précédemment. lambda_dynamo
Premier test
from __future__ import print_function
import os
from datetime import datetime
from urllib2 import urlopen
SITE = os.environ['site'] # URL of the site to check, stored in the site environment variable
EXPECTED = os.environ['expected'] # String expected to be on the page, stored in the expected environment variable
def validate(res):
return EXPECTED in res
def lambda_handler(event, context):
print('Checking {} at {}...'.format(SITE, event['time']))
try:
if not validate(urlopen(SITE).read()):
raise Exception('Validation failed')
except:
print('Check failed!')
raise
else:
print('Check passed!')
return event['time']
finally:
print('Check complete at {}'.format(str(datetime.now())))
Oui, j'ai une erreur.
print('Checking {} at {}...'.format(SITE, event['time']))
KeyError: 'time'
Jetons un coup d'œil aux variables utilisées par défaut avant de résoudre cette erreur.
Variables utilisées par défaut
- event - Les données d'événement définies par le test ou la source de déclenchement
- Context- Objet de contexte (Python)
- Les variables os.environ-Environment peuvent être définies sur AWS sans changer le code lui-même
event >> AWS Lambda utilise ce paramètre pour transmettre les données d'événement au gestionnaire. Ce paramètre est généralement un type de dict Python. Vous pouvez également utiliser les types list, str, int, float ou NoneType.
contexte >> AWS Lambda utilise ce paramètre pour fournir des informations d'exécution au gestionnaire. Ce paramètre sera de type LambdaContext.
Il est devenu.
Cette variable d'environnement peut être prise par os.environ
time ?
Nous allons maintenant gérer l'erreur précédente.
Ajoutez du temps pour les paramètres des événements de test.
{
"key3": "value3",
"key2": "value2",
"key1": "value1",
"time": "now...!"
}
Je ne pense pas qu'il y ait une erreur dans la partie Time.
START RequestId: a8708105-0948-11e7-b83e-b71ae2e4dbbe Version: $LATEST
Checking https://www.amazon.com/ at now...!...
Check failed!
Check complete at 2017-03-15 06:28:53.016209
HTTP Error 503: Service Unavailable: HTTPError
Traceback (most recent call last):
(réduction)
File "/usr/lib64/python2.7/urllib2.py", line 556, in http_error_default
raise HTTPError(req.get_full_url(), code, msg, hdrs, fp)
HTTPError: HTTP Error 503: Service Unavailable
END RequestId: a8708105-0948-11e7-b83e-b71ae2e4dbbe
REPORT RequestId: a8708105-0948-11e7-b83e-b71ae2e4dbbe Duration: 348.59 ms Billed Duration: 400 ms Memory Size: 128 MB Max Memory Used: 17 MB
Je reçois toujours une erreur, je vais donc faire différentes choses.
sauf que je gère des exceptions, donc j'obtiens une erreur!
Tout d'abord, je n'aime pas l'écran rouge, donc même si la demande d'url échoue, l'erreur disparaîtra.
try:
if not validate(urlopen(SITE).read()):
raise Exception('Validation failed')
except:
print('Check failed!')
raise
Dans le cas de Python, si rise est inséré à la fin, l'erreur sera renvoyée à l'instruction python pure comme après avoir détecté l'erreur avec except. Réécrivez ceci.
def lambda_handler(event, context):
print('Checking {} at {}...'.format(SITE, event['time']))
try:
if not validate(urlopen(SITE).read()):
raise Exception('Validation failed')
except Exception as e:
print('Check failed!')
print(e)
else:
print('Check passed!')
return event['time']
finally:
print('Check complete at {}'.format(str(datetime.now())))
Enfin, le chèque vert a réussi ...! !! !! !!
Attrapez l'erreur avec ʻexcept Exception comme e: Affiche une erreur avecprint (e) J'ai supprimérise` et je n'ai fait aucune erreur pour le moment.
J'ai fait beaucoup de recherches sur la gestion des erreurs de python ici. élever. ..
HTTP Error 503
Puisque l'erreur persiste, regardons la partie de Erreur HTTP 503: Service indisponible: Erreur HTTP.
De plus, comme il n'y a pas de mot La validation a échoué, validate (urlopen (SITE) .read ()) déterminera que l'erreur renvoyée dans cette partie est l'erreur ci-dessus.
def validate(res):
return EXPECTED in res
Cela ira à https://www.amazon.com/ et diagnostiquera s'il y a le mot «achats en ligne» dans le fichier html retourné.
Pour le moment, modifiez la variable d'environnement d'Amazon pour accéder à Google
- https://www.google.co.jp/
- Gmail
Puis
Checking https://www.google.co.jp/ at now...!...
Check passed!
Check complete at 2017-03-15 07:00:05.723916
Enfin chèque réussi! Est émis.
503 Service Unavailable Service non disponible. Le service est temporairement indisponible en raison d'une surcharge ou d'une maintenance. Par exemple, il est renvoyé lorsque l'accès est inondé et il devient impossible à traiter.
https://www.amazon.com/ Je ne peux pas tomber. J'ai oublié de renvoyer la variable d'environnement et de la vérifier du fond de mon cœur. Je l'ai changé en un simple service Google et cela a fonctionné. Refusez-vous l'accès de Lambda?
J'en suis accro.
indent
C'est une erreur Python, mais j'ai eu une erreur plusieurs fois même avec le nombre d'index.
Syntax error in module 'lambda_function':
unexpected indent (lambda_function.py, line 30)
J'ai étudié ici, mais c'est une guerre entre les onglets et les espaces (4). Tab vs Space War lors de l'écriture du code de programmation est enfin réglé Il est écrit que python est généralement écrit avec des espaces.
Le fichier que j'ai édité localement est inclus dans l'onglet Le code édité sur AWS peut être saisi avec 4 espaces. Je ne voulais pas voir cette erreur de retrait, alors je suis passé à l'école spatiale.
Requests
Vous pouvez utiliser urlopen tel quel, mais j'aimerais présenter des requêtes.
module python urllib2 Requêtes: HTTP pour les humains Code pour faire de même sans demandes
Il sera plus facile d'envoyer des messages tels que slack.
Codez maintenant.py
# coding: utf-8
from __future__ import print_function
import os
import json
import requests
from datetime import datetime
from urllib2 import urlopen
SITE = os.environ['site']
EXPECTED = os.environ['expected']
def validate(res):
return EXPECTED in res
def lambda_handler(event, context):
print('Checking {} at {}...'.format(SITE, event['time']))
try:
if not validate(urlopen(SITE).read()):
raise Exception('Validation failed')
except Exception as e:
print('Check failed!')
print(e)
else:
print('Check passed!')
finally:
print('Check complete at {}'.format(str(datetime.now())))
return "Finish"
Unable to import module 'lambda_function': No module named requests
Étant donné que les demandes sont un module externe, cette erreur se produit.
Utilisation de modules externes
J'avais l'habitude de mettre des modules Python non standard comme des requêtes, donc je vais l'écrire.
- À distance en utilisant lambda-uploader
- Zip localement vous-même
lambda-uploader Cette zone sera utile.
Développer, exécuter et déployer AWS Lambda à distance à l'aide de lambda-uploader Déployer AWS Lambda Python avec lambda-uploader
ZIP téléchargez votre code
Je l'ai fait avec ça. Vous pouvez le faire dès qu'il s'agit d'un seul coup. [AWS] Que faire lorsque vous souhaitez effectuer un pip avec Lambda
C'est comme pip install requests -t . dans votre dossier de travail et zippez-le.
Si vous pouvez télécharger avec succès, commencez à partir du précédent code actuel .py.
Slack
import
- urllib2
- requests
- L'un ou l'autre est acceptable
api
- Slack api (obtenez un jeton et allez sur xoxp-12131212 comme ça)
- Webhook (obtenir uniquement l'url)
- L'un ou l'autre est acceptable
Vous pouvez utiliser n'importe quelle combinaison, mais cette fois, nous utiliserons un simple webhook de requêtes *.
Cliquez ici pour obtenir l'URL du Webhook. Procédure d'acquisition d'URL Slack Webhook
def send_slack(text):
url = "L'URL qui était juste"
payload_dic = {
"text":text,
"icon_emoji":':grin:',
"channel":'bot-test2',
}
r = requests.post(url, data=json.dumps(payload_dic))
Vous pouvez envoyer Slack avec juste ça!
# coding: utf-8
from __future__ import print_function
import os
import json
import requests
from datetime import datetime
from urllib2 import urlopen
SITE = os.environ['site']
EXPECTED = os.environ['expected']
def validate(res):
return EXPECTED in res
def web_check():
try:
if not validate(urlopen(SITE).read()):
raise Exception('Validation failed')
except Exception as e:
print('Check failed!')
print(e)
else:
print('Check passed!')
finally:
print('Check complete at {}'.format(str(datetime.now())))
def lambda_handler(event, context):
print('Checking {} at {}...'.format(SITE, event['time']))
# web_check()
send_slack("test")
return "success!"
def send_slack(text):
url = "Ceci est mon URL"
payload_dic = {
"text":text,
"icon_emoji":':grin:',
"channel":'bot-test2',
}
r = requests.post(url, data=json.dumps(payload_dic))
J'ai déplacé le processus vers def web_check ():.
def lambda_handler(event, context):
print('Checking {} at {}...'.format(SITE, event['time']))
# web_check()
send_slack("test")
return "success!"
avec ça
Slack est arrivé.
Okay。 Ensuite, nous allons récupérer les données de DynamoDB.
DynamoDB
Structure de frais
Offre gratuite DynamoDB dans le cadre de l'offre gratuite AWS
| espace de rangement | Capacité d'écriture | Capacité d'écriture |
|---|---|---|
| 25GB | 25 | 25 |
Une unité de capacité gère une demande par seconde Capacité d'écriture de 1 unité: écrit jusqu'à 1 Ko de données une fois par seconde Capacité de lecture de 1 unité: jusqu'à 4 Ko de données peuvent être lues une fois par seconde 2,5 millions de requêtes de lecture à partir de flux DynamoDB sont disponibles gratuitement.
Pour le moment, si vous définissez la capacité d'écriture et la capacité de lecture à 1 lors de la création d'une base de données, je pense que cela ne coûtera pas d'argent s'il s'agit d'une trame libre. Cette fois, cela peut être environ une fois par minute, alors sélectionnez l'unité minimum de 1 (un niveau accessible une fois par seconde).
Essayer de faire
Pour le moment, les paramètres sont comme ça.
DynamoDB n'est accessible qu'avec la clé du hachage (type de dictionnaire, tableau associatif, etc. selon la langue). Cette fois, accédez avec id comme clé.
{
"id": 1,
"target": "Online Shopping",
"url": "https://www.amazon.com/"
}
{
"id": 2,
"target": "Gmail",
"url": "https://www.google.co.jp/"
}
Il est utilisé comme un tableau pour trouver la cible à partir de l'url.
Prenez les données précédentes du code python
#Ajouter deux
import boto3
from boto3.dynamodb.conditions import Key, Attr
def get_urls():
table = dynamodb.Table('sites')
response = table.scan()
sites = response["Items"]
return sites
J'ai ajouté une fonction et recherché la phrase cible à partir de l'url récupérée de DynamoDB avec get_urls ().
Courant.py
# coding: utf-8
from __future__ import print_function
import os
import json
import requests
from datetime import datetime
from urllib2 import urlopen
import boto3
from boto3.dynamodb.conditions import Key, Attr
dynamodb = boto3.resource('dynamodb')
def validate(res, target):
return target in res
def web_check(url, target):
print("Serching ... " + target)
try:
if validate(urlopen(url).read(), target):
print("Find!")
else:
print("Not Find!")
except Exception as e:
print('Error')
print(e)
def get_urls():
table = dynamodb.Table('sites')
response = table.scan()
sites = response["Items"]
return sites
def send_slack(text):
url = "https://hooks.slack.com/services/"
payload_dic = {
"text":text,
"icon_emoji":':grin:',
"channel":'bot-test2',
}
r = requests.post(url, data=json.dumps(payload_dic))
def lambda_handler(event, context):
print('Check start')
sites = get_urls()
for site in sites:
web_check(str(site["url"]), str(site["target"]))
return "success!"
Résultat de sortie
Check start
Serching ...Service de partage d'informations techniques sur https://qiita.com/
Find!
Serching ... Gmail at https://www.google.co.jp/
Find!
Serching ...Est-ce Gmail? à https://www.google.co.jp/
Not Find!
Serching ... Online Shopping at https://www.amazon.com/
Error
HTTP Error 503: Service Unavailable
END RequestId: 3992d81e-095e-11e7-b30a-1ddc7da9e992
J'ai aussi une erreur ici. Python trébuche!
'utf8' codec can't decode byte 0x90 in position 102: invalid start byte
Connaître Python UnicodeEncodeError
J'ai résolu l'erreur en me référant ici.
Lorsque type (site [" url "]) était terminé, c'était <type'unicode '>, donc
J'ai fait str (site [" url "]) et l'ai changé en <type'str '>.
Écrire DynamoDB à partir de Lambda
sites_check1 Ajouter une table
ajouter à
from datetime import datetime, timedelta
def insert(results):
date = datetime.now() + timedelta(hours=9)
id = 0
table = dynamodb.Table('sites_check')
table.put_item(
Item={
"id": id,
"date": date.strftime("%Y/%m/%d %H:%M"),
"result": results
}
)
Ajout de timedelta pour augmenter ou diminuer le temps. Titre ajouté à la table des sites DynamoDB. Code actuel
# coding: utf-8
from __future__ import print_function
import os
import json
import requests
from datetime import datetime, timedelta
from urllib2 import urlopen
import boto3
from boto3.dynamodb.conditions import Key, Attr
dynamodb = boto3.resource('dynamodb')
def validate(res, target):
return target in res
def web_check(url, target):
print("Serching ... " + target + " at " + url)
try:
if validate(urlopen(url).read(), target):
return "Find!"
else:
return "Not Find!"
except Exception as e:
return str(e)
def get_urls():
table = dynamodb.Table('sites')
response = table.scan()
sites = response["Items"]
return sites
def send_slack(text):
url = "https://hooks.slack.com/"
payload_dic = {
"text":text,
"icon_emoji":':grin:',
"channel":'bot-test2',
}
r = requests.post(url, data=json.dumps(payload_dic))
def insert(results):
date = datetime.now() + timedelta(hours=9)
id = 0
table = dynamodb.Table('sites_check')
table.put_item(
Item={
"id": id,
"date": date.strftime("%Y/%m/%d %H:%M"),
"result": results
}
)
def lambda_handler(event, context):
print('Check start')
results = {}
sites = get_urls()
for site in sites:
msg = web_check(str(site["url"]), str(site["target"]))
results[str(site["title"])] = msg
insert(results)
return "success!"
Voici le résultat de l'insertion des données.
{
"date": "2017/03/15 18:37",
"id": 0,
"result": {
"Amazon": "Find!", #Pour une raison quelconque, cela devient Find et aucune erreur ne se produit
"Google": "Find!",
"Google-2": "Not Find!",
"Qiita": "Find!"
}
}
{
"date": "2017/03/15 18:48",
"id": 0,
"result": {
"Amazon": "HTTP Error 503: Service Unavailable", #Str lorsqu'une erreur se produit ici(e)je l'ai fait
"Google": "Find!",
"Google-2": "Not Find!",
"Qiita": "Find!"
}
}
Si vous ne faites pas str (e), une erreur se produit car e n'est pas de type str.
Je suis habitué à Python, donc cela prend environ 10 minutes.
Je ne peux pas faire de commentaire sur json.
def get_recent_codes():
date = datetime.now() + timedelta(hours=9)
now = date.strftime("%Y/%m/%d %H:%M")
last = (date + timedelta(minutes=-9)).strftime("%Y/%m/%d %H:%M")
#Requête avec l'ID 0 et récupération des données dans les 10 minutes
response = table.query(
KeyConditionExpression=Key('id').eq(0) & Key('date').between(last, now)
)
return response
Le nombre de requêtes qui ont été récupérées est saisi dans response ['Count'].
Les données de la table extraites dans response ['Items'] sont incluses.
Si vous avez besoin d'autres données, veuillez retirer les données lors de l'impression, le cas échéant.
Le résultat ressemble à ceci
# coding: utf-8
from __future__ import print_function
import os
import json
import requests
from datetime import datetime, timedelta
from urllib2 import urlopen
import boto3
from boto3.dynamodb.conditions import Key, Attr
dynamodb = boto3.resource('dynamodb')
def validate(res, target):
return target in res
def web_check(url, target):
print("Serching ... " + target + " at " + url)
try:
if validate(urlopen(url).read(), target):
return "Find!"
else:
return "Not Find!"
except Exception as e:
return str(e)
def get_urls():
table = dynamodb.Table('sites')
response = table.scan()
sites = response["Items"]
return sites
def send_slack(text):
url = "https://hooks.slack.com/"
payload_dic = {
"text":text,
"icon_emoji":':grin:',
"channel":'bot-test2',
}
r = requests.post(url, data=json.dumps(payload_dic))
def insert(results):
table = dynamodb.Table('sites_check')
date = datetime.now() + timedelta(hours=9)
id = 0
table.put_item(
Item={
"id": id,
"date": date.strftime("%Y/%m/%d %H:%M"),
"result": results
}
)
def get_recent_codes():
table = dynamodb.Table('sites_check')
date = datetime.now() + timedelta(hours=9)
now = date.strftime("%Y/%m/%d %H:%M")
last = (date + timedelta(minutes=-9)).strftime("%Y/%m/%d %H:%M")
print(last + "De" + now + "Vérifiez jusqu'à")
response = table.query(
KeyConditionExpression=Key('id').eq(0) & Key('date').between(last, now)
)
return response
def lambda_handler(event, context):
print('Check start')
results = {}
sites = get_urls()
for site in sites:
msg = web_check(str(site["url"]), str(site["target"]))
results[str(site["title"])] = msg
insert(results)
print(get_recent_codes())
return "success!"
Il semble que vous puissiez le faire fonctionner régulièrement et faire diverses choses.
à la fin
Je suis un amateur de Python et de Lambda, j'ai donc fait beaucoup d'erreurs. Cependant, je pense que les architectes avec Python et Lambda seront très utiles, alors j'aimerais continuer à les utiliser.
Recommended Posts