fast-api-boilerplate.py

fast_api_boilerplate.py

import json
from fastapi import FastAPI
import datetime
import boto3
app = FastAPI()

def get_int_timestamp():
    return int(datetime.datetime.now().timestamp())

def dev_get_bearer_token():
    payload = {
        "username": "16411624-7",
        "password": "1716K727ays!"

        }
    creds = json.loads(open("cred_aws.json").read())
    lambda_client = boto3.client('lambda',region_name="us-east-1", aws_access_key_id=creds.get("AWS_ACCESS_KEY_ID"), aws_secret_access_key=creds.get("AWS_SECRET_ACCESS_KEY"))
    response  = lambda_client.invoke(FunctionName="dt_token_generator", InvocationType="RequestResponse", Payload=json.dumps(payload))
    response_json = json.loads(response.get("Payload").read().decode("utf-8"))
    dict_bearer_token = json.loads(response_json.get("body"))
    return dict_bearer_token
    
def refresh_token():
    new_bearer = dev_get_bearer_token()
    timestamp = get_int_timestamp()
    bearer = new_bearer.get("bearer_token")
    new_fresh_token = {"bearer": bearer, "timestamp": timestamp}
    with open("refresh_token.json", "w") as f:
        f.write(json.dumps(new_fresh_token))
    return new_fresh_token

def get_fresh_token():
    fresh_token = json.loads(open("fresh_token.json").read())
    if fresh_token["timestamp"] == "":
        return refresh_token()
    if fresh_token["timestamp"] + 6700 < get_int_timestamp():
        return refresh_token()
    return fresh_token
    



# Spagetti code! lo consume scrapy de vuelta
# TODO: En prd, debe estar desacoplado
@app.get("/give_me_a_token")
def give_me_a_token():
    return get_fresh_token()

if __name__ == "__main__":
    import uvicorn

    uvicorn.run(app, host="0.0.0.0", port=8000)