当前位置：首页 > news >正文

testuserpython

news 2026/3/26 15:36:41

import json
import boto3
import flask
import psycopg2
from flask import request

app = flask.Flask(name)

@app.route('/')
def route_root():
return flask.jsonify('OK')

dynamodb = boto3.client('dynamodb')

@app.route('/post_data1', methods=['POST'])
def route_data1():
data = json.loads(request.data)
dynamodb.put_item(
TableName="dynamodb-table",
Item = {
'id': {'S': str(data['id'])},
'value': {'S': str(data['value'])}
}
)
return flask.jsonify('OK')

def get_conn(): # TODO: fill this
return psycopg2.connect(
host = '',
user = '',
password = '',
database = ''
)

@app.route('/post_data2', methods=['POST'])
def route_data2():
data = json.loads(request.data)
conn = get_conn()
cursor = conn.cursor()
cursor.execute('INSERT INTO cloudraiser (id, value) VALUES (%s, %s)', (str(data['id']), str(data['value'])))
conn.commit()
return flask.jsonify('OK')

@app.route('/init_db')
def route_init_db():
conn = get_conn()
cursor = conn.cursor()
cursor.execute('CREATE TABLE cloudraiser (id VARCHAR, value VARCHAR);')
conn.commit()
return flask.jsonify('OK')

@app.route('/get_value')
def route_get():
id = request.args.get('id')
data1 = dynamodb.query(
TableName='ddb-table',
KeyConditionExpression='#id = :id',
ExpressionAttributeNames= {
'#id': 'id'
},
ExpressionAttributeValues= {
':id': {'S': str(id)}
}
)['Items'][0]['value']['S']
conn = get_conn()
cursor = conn.cursor()
cursor.execute('SELECT value FROM cloudraiser WHERE id = %s', (str(id), ))
data2 = cursor.fetchone()[0]
conn.commit()
return flask.jsonify({
'message': str(int(data1) + int(data2))
})