def register_action()

in dialogflow-cx/vpc-sc-demo/backend/analytics_utilities.py [0:0]

• 38 lines of code
• 6 McCabe index (conditional complexity)

def register_action(request, response, action, data=None):
    """Register a user action with the analytics database."""
    target_database = "prod" if su.is_prod() else "dev"
    if data is None:
        data = {}

    session_id = request.cookies.get("session_id")
    if session_id:
        tok_response = get_token.get_token(request, token_type="email")
        if "response" in tok_response:
            logging.error(
                "Error retriving user information: %s", tok_response["response"]
            )
            response.delete_cookie("session_id", domain=su.user_service_domain(request))
            response.delete_cookie(
                "user_logged_in", domain=su.user_service_domain(request)
            )
            user_hash = None
        else:
            user_hash = hashlib.md5(tok_response["email"].encode("utf-8")).hexdigest()
    else:
        user_hash = None
    instance = {
        "visit_id": request.cookies.get("visit_id"),
        "session_id": session_id,
        "user_hash": user_hash,
        "action": action.value,
        "timestamp": int(datetime.datetime.now().timestamp()),
        "action_data": data,
    }
    validate_data(instance)

    client = bigquery.Client(credentials=asu.get_credentials())
    table = client.get_table(
        client.dataset(os.environ["ANALYTICS_DATABASE"]).table(target_database)
    )
    insert_response = client.insert_rows_json(json_rows=[instance], table=table)
    if insert_response:
        logging.error("Error inserting into analytics database: %s", insert_response)

    update_visit_cookie(request, response)
    return response