from pynamodb.models import Model
from pynamodb.attributes import (
UnicodeAttribute,
NumberAttribute,
ListAttribute,
)
class UserModel(Model):
class Meta:
table_name = "USERS"
region = 'ap-northeast-1'
id = UnicodeAttribute(hash_key=True)
name = UnicodeAttribute(null=False)
age = NumberAttribute(null=False)
hobbies = ListAttribute(null=True)
CRUD APIの実装
from chalice import Chalice
from chalice import (
ChaliceViewError,
NotFoundError,
)
from chalicelib.UserModel import UserModel
import logging
import json
import uuid
logger = logging.getLogger()
logger.setLevel(logging.INFO)
app = Chalice(app_name='pynamodb-crud-api')
@app.route('/user', methods=['POST'],
content_types=['application/json'])
def create_user():
request = app.current_request
logger.info(json.dumps(request.json_body))
try:
if not UserModel.exists():
UserModel.create_table(
read_capacity_units=1, write_capacity_units=1, wait=True)
id = str(uuid.uuid4())
user = UserModel(id)
user.name = request.json_body['name']
user.age = int(request.json_body['age'])
if 'hobbies' in request.json_body:
user.hobbies = request.json_body['hobbies']
user.save()
except Exception as e:
logger.error(e)
raise ChaliceViewError(str(e))
return {
'id': user.id,
'name': user.name,
'age': user.age,
'hobbies': user.hobbies,
}
@app.route('/users', methods=['GET'])
def get_users():
try:
users = UserModel.scan()
response = []
for user in users:
response.append({
'id': user.id,
'name': user.name,
'age': user.age,
'hobbies': user.hobbies,
})
except Exception as e:
logger.error(e)
raise ChaliceViewError(str(e))
return response
@app.route('/user/{id}', methods=['GET'])
def get_user_by_id(id):
try:
user = UserModel.get(id)
except UserModel.DoesNotExist:
logger.warn(id + ' is not found.')
raise NotFoundError(id + ' is not found.')
except Exception as e:
logger.error(e)
raise ChaliceViewError(str(e))
return {
'id': user.id,
'name': user.name,
'age': user.age,
'hobbies': user.hobbies,
}
@app.route('/user/{id}', methods=['PUT'],
content_types=['application/json'])
def update_user_by_id(id):
request = app.current_request
logger.info(json.dumps(request.json_body))
name = request.json_body['name']
age = int(request.json_body['age'])
if 'hobbies' in request.json_body:
hobbies = request.json_body['hobbies']
else:
hobbies = None
try:
user = UserModel.get(id)
user.update(actions=[
UserModel.name.set(name),
UserModel.age.set(age),
UserModel.hobbies.set(hobbies),
])
except UserModel.DoesNotExist:
logger.warn(id + ' is not found.')
raise NotFoundError(id + ' is not found.')
except Exception as e:
logger.error(e)
raise ChaliceViewError(str(e))
return {
'id': user.id,
'name': user.name,
'age': user.age,
'hobbies': user.hobbies,
}
@app.route('/user/{id}', methods=['DELETE'])
def delete_user_by_id(id):
try:
user = UserModel.get(id)
user.delete()
except UserModel.DoesNotExist:
logger.warn(id + ' is not found.')
raise NotFoundError(id + ' is not found.')
except Exception as e:
logger.error(e)
raise ChaliceViewError(str(e))
return {
'id': user.id,
'name': user.name,
'age': user.age,
'hobbies': user.hobbies,
}
ディスカッション
コメント一覧
まだ、コメントがありません