mirror of
https://github.com/projecthorus/sondehub-infra.git
synced 2024-12-18 20:57:56 +00:00
add postgres prototype
This commit is contained in:
Michaela
parent
b3d0b84465
commit
4a4f0d5402
1
postgres_insert/README
Normal file
1
postgres_insert/README
Normal file
@ -0,0 +1 @@
|
||||
This isn't used yet.
|
||||
109
postgres_insert/insert-batch.py
Normal file
109
postgres_insert/insert-batch.py
Normal file
@ -0,0 +1,109 @@
|
||||
import logging
|
||||
from urllib import parse
|
||||
import boto3
|
||||
from botocore.exceptions import ClientError
|
||||
import json
|
||||
logger = logging.getLogger(__name__)
|
||||
logger.setLevel('DEBUG')
|
||||
|
||||
from botocore import UNSIGNED
|
||||
from botocore.config import Config
|
||||
|
||||
s3 = boto3.client('s3', config=Config(signature_version=UNSIGNED))
|
||||
|
||||
import psycopg2
|
||||
import psycopg2.extras
|
||||
import os
|
||||
|
||||
DB_HOST = os.getenv("DB_HOST")
|
||||
DB_USER = os.getenv("DB_USER")
|
||||
DB_PASSWORD = os.getenv("DB_PASSWORD")
|
||||
DB_DATABASE = os.getenv("DB_DATABASE")
|
||||
|
||||
con = psycopg2.connect(database=DB_DATABASE, user=DB_USER,
|
||||
password=DB_PASSWORD, host=DB_HOST, port="5432")
|
||||
|
||||
|
||||
def lambda_handler(event, context):
|
||||
# Parse job parameters from Amazon S3 batch operations
|
||||
invocation_id = event['invocationId']
|
||||
invocation_schema_version = event['invocationSchemaVersion']
|
||||
|
||||
results = []
|
||||
result_code = None
|
||||
result_string = None
|
||||
|
||||
for task in event['tasks']:
|
||||
task_id = task['taskId']
|
||||
|
||||
try:
|
||||
obj_key = parse.unquote(task['s3Key'], encoding='utf-8')
|
||||
bucket_name = task['s3BucketArn'].split(':')[-1]
|
||||
|
||||
logger.info("Got task: %s.", obj_key)
|
||||
|
||||
response = s3.get_object(
|
||||
Bucket=bucket_name, Key=obj_key
|
||||
)
|
||||
|
||||
payload = json.loads(response["Body"].read())
|
||||
|
||||
try:
|
||||
cur = con.cursor(cursor_factory=psycopg2.extras.DictCursor)
|
||||
cur.execute(
|
||||
"""
|
||||
INSERT INTO telemetry (datetime, serial, type, uploader_callsign, frame, frame_data, "position")
|
||||
VALUES (%s, %s, %s, %s, %s, %s, ST_SetSRID(ST_MakePoint(%s, %s, %s), 4326)::Point);
|
||||
""",
|
||||
(payload["datetime"], payload["serial"], payload["type"], payload["uploader_callsign"], payload["frame"],json.dumps(payload), payload["lat"], payload["lon"], payload["alt"] )
|
||||
)
|
||||
con.commit()
|
||||
result_code = 'Succeeded'
|
||||
result_string = f"Successfully inserted into DB" \
|
||||
f" for object {obj_key}."
|
||||
logger.info(result_string)
|
||||
except:
|
||||
result_code = 'TemporaryFailure'
|
||||
result_string = f"Attempt to insert " \
|
||||
f"{obj_key}"
|
||||
logger.info(result_string)
|
||||
except Exception as error:
|
||||
# Mark all other exceptions as permanent failures.
|
||||
result_code = 'PermanentFailure'
|
||||
result_string = str(error)
|
||||
logger.exception(error)
|
||||
finally:
|
||||
results.append({
|
||||
'taskId': task_id,
|
||||
'resultCode': result_code,
|
||||
'resultString': result_string
|
||||
})
|
||||
return {
|
||||
'invocationSchemaVersion': invocation_schema_version,
|
||||
'treatMissingKeysAs': 'PermanentFailure',
|
||||
'invocationId': invocation_id,
|
||||
'results': results
|
||||
}
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
print(lambda_handler(
|
||||
|
||||
{
|
||||
"invocationSchemaVersion": "1.0",
|
||||
"invocationId": "YXNkbGZqYWRmaiBhc2RmdW9hZHNmZGpmaGFzbGtkaGZza2RmaAo",
|
||||
"job": {
|
||||
"id": "f3cc4f60-61f6-4a2b-8a21-d07600c373ce"
|
||||
},
|
||||
"tasks": [
|
||||
{
|
||||
"taskId": "dGFza2lkZ29lc2hlcmUK",
|
||||
"s3Key": "date/1253-02-09T11:12:58.000000Z-17052971-982805f8-fa15-4183-b4a2-f1991b6d8f59.json",
|
||||
"s3VersionId": "1",
|
||||
"s3BucketArn": "arn:aws:s3:us-east-1:0123456788:sondehub-open-data"
|
||||
}
|
||||
]
|
||||
}, {}
|
||||
)
|
||||
)
|
||||
|
||||
60
postgres_insert/insert.py
Normal file
60
postgres_insert/insert.py
Normal file
@ -0,0 +1,60 @@
|
||||
import psycopg2
|
||||
import psycopg2.extras
|
||||
import os
|
||||
import json
|
||||
|
||||
DB_HOST = os.getenv("DB_HOST")
|
||||
DB_USER = os.getenv("DB_USER")
|
||||
DB_PASSWORD = os.getenv("DB_PASSWORD")
|
||||
DB_DATABASE = os.getenv("DB_DATABASE")
|
||||
|
||||
con = psycopg2.connect(database=DB_DATABASE, user=DB_USER,
|
||||
password=DB_PASSWORD, host=DB_HOST, port="5432")
|
||||
|
||||
|
||||
def main(event, context):
|
||||
cur = con.cursor(cursor_factory=psycopg2.extras.DictCursor)
|
||||
cur.execute(
|
||||
"""
|
||||
INSERT INTO telemetry (datetime, serial, type, uploader_callsign, frame, frame_data, "position")
|
||||
VALUES (%s, %s, %s, %s, %s, %s, ST_SetSRID(ST_MakePoint(%s, %s, %s), 4326)::Point);
|
||||
""",
|
||||
(event["datetime"], event["serial"], event["type"], event["uploader_callsign"], event["frame"],json.dumps(event), event["lat"], event["lon"], event["alt"] )
|
||||
)
|
||||
con.commit()
|
||||
cur.close()
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main(
|
||||
{
|
||||
"software_name": "radiosonde_auto_rx",
|
||||
"software_version": "1.5.1",
|
||||
"uploader_callsign": "F4ICT-F4KLR",
|
||||
"uploader_position": "50.4977572,2.8616905555555556",
|
||||
"uploader_antenna": "Diamond X-200",
|
||||
"time_received": "2021-04-09T01:18:01.170006Z",
|
||||
"datetime": "2021-04-09T01:18:17.001000Z",
|
||||
"manufacturer": "Vaisala",
|
||||
"type": "RS41",
|
||||
"serial": "S2840432",
|
||||
"subtype": "RS41-SG",
|
||||
"frame": 8894,
|
||||
"lat": 51.05502,
|
||||
"lon": 2.5822,
|
||||
"alt": 949.41797,
|
||||
"temp": 2.1,
|
||||
"humidity": 63.8,
|
||||
"vel_v": -7.69568,
|
||||
"vel_h": 7.69487,
|
||||
"heading": 83.04947,
|
||||
"sats": 7, "batt": 2.6,
|
||||
"frequency": 404.801,
|
||||
"burst_timer": 28737,
|
||||
"snr": 19.3,
|
||||
"user-agent": "Amazon CloudFront",
|
||||
"position": "51.05502,2.5822",
|
||||
"upload_time_delta": -0.745924,
|
||||
"uploader_alt": 21.0
|
||||
}, {}
|
||||
)
|
||||
@ -4,8 +4,6 @@ import zlib
|
||||
import base64
|
||||
import datetime
|
||||
import functools
|
||||
from aws_xray_sdk.core import xray_recorder
|
||||
from aws_xray_sdk.core import patch_all
|
||||
from awscrt import io, mqtt, auth, http
|
||||
from awscrt.exceptions import AwsCrtError
|
||||
from awsiot import mqtt_connection_builder
|
||||
@ -32,16 +30,11 @@ import os
|
||||
# work out how to have a dev env
|
||||
|
||||
|
||||
patch_all()
|
||||
|
||||
|
||||
event_loop_group = io.EventLoopGroup(1)
|
||||
host_resolver = io.DefaultHostResolver(event_loop_group)
|
||||
|
||||
|
||||
io.init_logging(io.LogLevel.Error, "stderr")
|
||||
|
||||
|
||||
def connect():
|
||||
global connect_future, mqtt_connection
|
||||
session = boto3.session.Session()
|
||||
@ -91,7 +84,7 @@ def lambda_handler(event, context):
|
||||
except:
|
||||
pass
|
||||
payload = json.loads(event["body"])
|
||||
|
||||
print(payload)
|
||||
tasks = []
|
||||
first = False
|
||||
if "user-agent" in event["headers"]:
|
||||
@ -121,46 +114,3 @@ def lambda_handler(event, context):
|
||||
|
||||
return {"statusCode": 200, "body": "^v^ telm logged"}
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
payload = {
|
||||
"version": "2.0",
|
||||
"routeKey": "PUT /sondes/telemetry",
|
||||
"rawPath": "/sondes/telemetry",
|
||||
"rawQueryString": "",
|
||||
"headers": {
|
||||
"accept": "*/*",
|
||||
"accept-encoding": "gzip, deflate",
|
||||
"content-encoding": "gzip",
|
||||
"content-length": "2135",
|
||||
"content-type": "application/json",
|
||||
"host": "api.v2.sondehub.org",
|
||||
"user-agent": "autorx-1.4.1-beta4",
|
||||
"x-amzn-trace-id": "Root=1-6015f571-6aef2e73165042d53fcc317a",
|
||||
"x-forwarded-for": "103.107.130.22",
|
||||
"x-forwarded-port": "443",
|
||||
"x-forwarded-proto": "https",
|
||||
"date": "Sun, 31 Jan 2021 00:21:45 GMT",
|
||||
},
|
||||
"requestContext": {
|
||||
"accountId": "143841941773",
|
||||
"apiId": "r03szwwq41",
|
||||
"domainName": "api.v2.sondehub.org",
|
||||
"domainPrefix": "api",
|
||||
"http": {
|
||||
"method": "PUT",
|
||||
"path": "/sondes/telemetry",
|
||||
"protocol": "HTTP/1.1",
|
||||
"sourceIp": "103.107.130.22",
|
||||
"userAgent": "autorx-1.4.1-beta4",
|
||||
},
|
||||
"requestId": "Z_NJvh0RoAMEJaw=",
|
||||
"routeKey": "PUT /sondes/telemetry",
|
||||
"stage": "$default",
|
||||
"time": "31/Jan/2021:00:10:25 +0000",
|
||||
"timeEpoch": 1612051825409,
|
||||
},
|
||||
"body": "H4sIAHD1FWAC/8WbX2/byBXFv0rg5/Vg7v87eduHYh8LJNui2KIwlFjZGHDsVJbTLop+996hszVHXpECKI7hJ1GURiR/vnPOmTt//8/Fw+OH/W9ftxdv31y8e89w+f6nix/eXOy3X77GoUvDVOLll83d46fNx/3jbrurZ/51c/Owud3UMx+2u5vNbT34HpAyKNSjt5t9/ThxMs/OceTTbvOljsLKVj92V78IOEG8+La9vfpW30uGkOvZ15v9dn8zfOACM8JlhkuCn3N+C/lt5pQz5Jx/qUN92OzrWJiGr73/tP/XZre9unsa7WK3ub65f7i/u95ebR7391e7f18MP2b7z8ft3cff6qAZkuT6Mz5v4+S7X+OYSyoqWr/x8+OXm+ubfT3Tk8aBzW0dDoyLpWwu/v0CPtejORURpTj0+PX2fnO93V193NzePtz8eld/zY/v/vTT1Y9/+fnPV+/+Nty8zf4hjtdxbu/rGUAlgZfSXMu37e7hZnj7AlLcscsP2/2Gh2t/3D3sr+qNqndTRUjqwxs9z+FhxvtXu+3H7c237fXLO5rLW/YU912h/HLx3x/e9IGCWij8GBSe4lYjzUIhq0OhiXPJ5RAKG0EhmVKxLDSGAhI+XcACJkoeBu7JRIwKQCKvx0R5ZsIbJiiRGJRZJvQPmfCzMuGifIiENEiU5BnxsE4UEF3KBEJfJiQHE04Mr8aE5GNMlFQgyvEsE7Z2nSg5sRr4JBQQaAoUaKFwRFgMBVFnKKK+ocSTOhUKOzsU8AwFDpP071BgEmHzWSi8AxQU8/oLKHgMBWIoipg9DiaPbAWXQsHSGQpMQKgsrwcFHlMUFrzSvMosqzMR0xiY6SET1DBR65UXbpmIp4plKRPSWWUK1d+tjKcyoWdngsZMeKMoOP79ZlVm1Oj1oWBGmYaCIGRHztbOHiI2TCiLoNDOMlPiW9lyplOhkHP7UeFnKKSBIlwgC8/KzPh/XFlmFklmBodMQMNESFF4AmesKNSGB7OICe+MhCQQRuSOSEiLhDwjYYlGSGgq7jw7eYQMWR8JRi0vIgocM8FRccPClXbyIAspspSJQZL0hEJTPDgi6AiFtlDoMSh4SE9wFgpaffIIaUMmk3WCPVm2gzIhmGFpQvGU3PVEwgIJx9ONxxmQ8BaJUZQZhq5JrbyIzk8dq0eZpcY4eRDDx5moFRcdxqlVSTEx0lI5AdDbd8SVGZeipzLBy5koLRM+ZmLsO6BO4jTPxOpJZgm7W1ym64TmhOrSMmEU9WUpE6idmQgn52Deceo4QKIcQ8KTepbZ0Ap0bSQgjqH44IAmmAgdKiZj3xGFjtmWhlZAPZmo7j4ncA+c+9WJQTE9Q6F5bDu08aI59O+8xrS1NSbkkLEiRSdFpoX9CbVeGiiIoggvhaJraFWhgBhVXE42HrQcijagUBhXCm6MB0fN1lkofP1KkWveizpZKSy8s/FBpYjJAxdXCuHOUGCKY+X0NY8zQNG6UcVjUIQADmDnK8X6UWZMayAEk6mVx5AAhg0TJX4WL2VCqTMT4atzMT959sjLmWjNqDZJprZNFApoc0xg7mBGTc18sk547TxQ54YJycSLJw/DzkzEt4LF32lMwBlWPA6ZOBpkhsOnImU2oUBYHYowNwh0sGJuw0Lx/6GIk0ifnNJIZpaSly6DgfeGYjDWRfVUKHA5FNZCIcegqKtPJc8qCsQOTDDyYaGwpttK43PsTZJZEokPfUOLkCi9kYipkjTu/KlIwHIk2tRKdYwEtamVUpnVE0irp9tcF7helokWiZA/XA4SiuIOS5nA3JuJeiXop/bf5TP01ZQ2otBRkknJ2haKKGHzZWL9JJMTm2WbZAJiyLCs2qZWhrqYCeitMb3Szcqvx4SP60RpzGgIt+HxTDMhHZhwwMPU6pAJC+/s1K540BmSTMTeXjRGFYWOTEBLxHOQGf5Bm6XyzEizjVaoHZwomTEcEjHu0tU6gRk5tkjE1crSBQ+kzplV/O6wv9mlHxK5hcLycSiETWebatA6lImiXnAaijCe8tT0P4Iii+JS24HcOd2uHWKWKXesE7lNtw2OLXmEbOenNbNpKHx1jQmpEM9AQcGwFm71BIdxWqwnunbfPbWkQAihUxut1mACjzFBdY8HzxeK9XPMuPmWdVpPMIToKGgNE8IkSzsoUHszEfNg+KxTV8vXYOJoR2ZJITfybGRFq+eYbuGKRWGaCU+Y2ds6kUNRL22+Q/POTISRK3El8npMNDmmNF7098bnaSagwwZBA+LpyEpqwiYibYyZwRYz4b3rhATdYJxPZcLPLjLl2NxhQ2Y1iwR2QCKTwAs5IQ0SlgpIaZc7yCkvXRbF0hsJDSQcvfRDogVCjwGBCUV4tqeGqMPmwO+txMeB0BjRm7gqSgQuXxJ9IqonD3UtmuPjp/JgZ+bBjvEQr7DML4gSr85DlKqC05m2Wt035QdLX7WRcikQ0BsIj/sO+fWA8GObAuu/peNsok3SQVkWIPBJB2pUUxRop4zamLnUgRL2jqpiVDSR0o2Iw0TbyniVg5uuCTHKs/klaQcHGhdSaJqJUnMpxdZt+FOn3iImeseXmKPexaj8SlXC89h/SuM/UVFn1zjIOiTaHFcCUy0TUQ/CWZC0mYTL8u3k1LsJE7/vbj3Zf+q5vYbDMSbqLlHP89LSO+wSrcvUNslEwbpaxN7mVLp4PyBJZymBcSGs2TtKiUMkms3kuSkTwjq/m5xKhzUOLPzSbrRIeG21Odjno8ulhPYmIvgXLIAdi0QbUjkdKxKYxGG+KZdzByKgiL3Y0pFHRFhtW6TcJBIlFV7uN6xzIIHxrSGA2AKJf/wPWvB/4NlMAAA=",
|
||||
"isBase64Encoded": True,
|
||||
}
|
||||
lambda_handler(payload, {})
|
||||
|
||||
Loading…
Reference in New Issue
Block a user