¿Por qué fastAPI no está creando la base de datos y devolviendo el resultado aquí?

15 Jun, 2023 Programación 0

Estoy tratando de crear un sistema de puntos para tweets. He transmitido tweets con #Python a una base de datos MySQL y estoy tratando de crear un sistema de puntos para lo mismo.

from typing_extensions import Self
import requests
import os
import json
import mysql.connector
from mysql.connector import Error
bearer_token = “&$Bearer”#Obtener tweet ids de un usuario especificado de la base de datos
ids=[]
class tweet_id:
def init(self, name):
self.name = name
def get_tweet_ids(self, name):
try:
connection = mysql.connector.connect(host=’localhost’,
database=’twitterdb’,
user=’root’,
password=’contraseña@123′)

        cursor = connection.cursor()
        sql_select_query = """SELECT tweetid FROM twitterdb.StreamData WHERE username = %s"""
        # set variable in query
        cursor.execute(sql_select_query, (name,))
        # fetch result
        record = cursor.fetchall()

        for row in record:
            ids.append(int(row[0]))

    except mysql.connector.Error as error:
        print("Error al obtener el registro de la tabla MySQL: {}".format(error))

    """finally:
        if connection.is_connected():
            cursor.close()
            connection.close()"""

def create_url():
tweet_fields = “tweet.fields=public_metrics”
converted_list = [str(element) for element in ids]
id_list = “,”.join(converted_list)
url = “https://api.twitter.com/2/tweets?ids={}&{}&”.format(id_list, tweet_fields)
return url

def bearer_oauth(r):
“””
Método requerido por la autenticación de tokens de portador.
“””

r.headers["Authorization"] = f"Bearer {bearer_token}"
return r

def connect_to_endpoint(url):
response = requests.request(“GET”, url, auth=bearer_oauth)
print(response.status_code)
if response.status_code != 200:
raise Exception(
“La solicitud devolvió un error: {} {} {}”.format(
response.status_code, response.text, ids
)
)
return url
return response.json()

def main():
#def init(connect, append_to_database):
#Self.connect = connect
#Self.append_to_database = append_to_database
def connect(tweetid, retweet_count, reply_count, like_count, quote_count):
“””
Conectar a la base de datos MySQL e insertar datos de Twitter
“””
try:
con = mysql.connector.connect(host=’localhost’,
database=’twitterdb’, user=’root’, password=’contraseña@123′, charset=’utf8′)

        if con.is_connected():
            """
            Insertar datos de Twitter
            """
            cursor = con.cursor(buffered=True)
            # twitter, golf
            query = "INSERT INTO Metrics (tweetid,retweet_count,reply_count,like_count,quote_count) VALUES (%s, %s, %s, %s, %s)"
            cursor.execute(query, (tweetid,retweet_count,reply_count,like_count,quote_count))
            con.commit()


    except Error as e:
        print(e)

    cursor.close()
    con.close()

    return
def append_to_database(json_response):
#Loop through each tweet
    for tweet in json_response['data']:  
        # Tweet ID
        tweetid = tweet['id']
        # Tweet metrics
        retweet_count = tweet['public_metrics']['retweet_count']
        reply_count = tweet['public_metrics']['reply_count']
        like_count = tweet['public_metrics']['like_count']
        quote_count = tweet['public_metrics']['quote_count']
        connect(tweetid, retweet_count, reply_count, like_count, quote_count)
    url = create_url()
    json_response = connect_to_endpoint(url)
    append_to_database(json_response)

#Función para conectar e insertar en la base de datos

#Función para calcular la suma de puntos y mostrarla
class summer:
like_points=0
reply_points=0
total_rts=0
rt_points=0
total=0
def sum_fun():

    try:
        con = mysql.connector.connect(host='localhost',
        database='twitterdb', user='root', password='contraseña@123', charset='utf8')

        if con.is_connected():
            cursor = con.cursor(buffered=True)
            def sum_rts():
                cursor.execute("SELECT SUM(retweet_count) FROM twitterdb.Metrics")
                sum1=cursor.fetchall()[0][0]
                if sum1 is None:
                    return 0;
                else:
                    return int(sum1)
            def sum_replies():
                cursor.execute("SELECT SUM(reply_count) FROM twitterdb.Metrics")
                sum2=cursor.fetchall()[0][0]
                if sum2 is None:
                    return 0
                else:
                    return int(sum2)
            def sum_likes():
                cursor.execute("SELECT SUM(like_count) FROM twitterdb.Metrics")
                sum3=cursor.fetchall()[0][0]
                if sum3 is None:
                    return 0
                else:
                    return int(sum3)
            def sum_qts():
                cursor.execute("SELECT SUM(quote_count) FROM twitterdb.Metrics")
                sum4=cursor.fetchall()[0][0]
                if sum4 is None:
                    return 0
                else:
                    return int(sum4)      
            like_points= (20*(sum_likes()))
            reply_points=  (100 * (sum_replies()))
            total_rts= (sum_rts() + sum_qts())
            rt_points = (300 * total_rts)
            total = (like_points + reply_points + rt_points)

            return total

            #print("Puntos de Me gusta:", like_points)
            #print("Puntos de respuesta:", reply_points)
            #print("Puntos de retuit:", rt_points)
            #print("Puntos totales:",total)
        # print(points)



    except Error as e:
        print(e)

    cursor.close()
    con.close()

def clear():
“””
Conectar a la base de datos MySQL e insertar datos de Twitter
“””
try:
con = mysql.connector.connect(host=’localhost’,
database=’twitterdb’, user=’root’, password=’Mysql@123′, charset=’utf8′)

    if con.is_connected():
        cursor = con.cursor(buffered=True)
        cursor.execute("truncate table twitterdb.Metrics")




except Error as e:
    print(e)

#cursor.close()
#con.close()

return

Además, he creado una API en FastAPI para activar todas las funcionalidades en el script anterior y obtener las salidas,como puntos,reply_points,rt_points y total enviados a través de una API. La API acepta el valor username mediante una solicitud POST y activa el script.

API code:

from fastapi import FastAPI
from pydantic import BaseModel
from metrics import tweet_id
from metrics import create_urls
from metrics import summer
import metrics
from typing_extensions import Self

app = FastAPI()

class Username(BaseModel):
username:str

@app.post(‘/Username’)
def Username(Username : Username):
username=Username.username
tweets_list = tweet_id(username)
tweets_list.get_tweet_ids(str(username))
metrics.clear()
metrics.main()
points=summer.sum_fun()
return{points.total}

if name == “main“:
uvicorn.run(“api:app”, host=”127.0.0.1″, port=5000, log_level=”info”)

No puedo obtener la salida y aunque la solicitud está completa, obtengo “null” como resultado. ¿Por qué está sucediendo esto? Además, soy muy nuevo en todo esto, así que las sugerencias y modificaciones para mejorar el código son muy bienvenidas. Gracias.

Tags: AZURE-DATABASE-MYSQL, FASTAPI, MYSQL-CONNECTOR, MYSQL-CONNECTOR-PYTHON, PYTHON-3.X

Answer

davy.ai

June 15, 2023 at 9:42 pm
Basándose en los fragmentos de código proporcionados, parece que hay algunos problemas:
1. En el código de la API, las declaraciones de importación para tweet_id, create_urls y summer no deberían tener la extensión .py. Reemplace import metrics con from metrics import tweet_id, create_urls, summer.
2. En el código de la API, la declaración de devolución debe ser return {'points': points}, donde points es el total calculado por summer.sum_fun().
3. En la función connect de la función main, no hay manejo de errores cuando el identificador de tweet ya existe en la base de datos. Es posible que desee agregar una verificación para esto y actualizar la entrada en lugar de insertar una nueva si el identificador de tweet ya existe.
4. En la función connect_to_endpoint, la declaración de return url debe estar fuera del bloque if para asegurarse de que la URL siempre se devuelva.
5. En la clase summer, el método sum_fun debería ser un método estático ya que no se basa en ninguna variable de instancia. Agregue @ staticmethod antes de la definición del método.
6. En la clase summer, las variables like_points, reply_points, total_rts, rt_points y total deberían definirse como variables de instancia en lugar de variables de clase. Cambie class summer: a class Summer: y defina las variables dentro del método __init__ en lugar de fuera de él. Por ejemplo, self.like_points = 0.
7. En la función clear, las declaraciones cursor.close () y con.close () deberían eliminarse los comentarios para cerrar la conexión de la base de datos correctamente.
8. Es posible que desee agregar un manejo de errores más detallado en todo el script para detectar y manejar errores potenciales de una manera más informativa. Por ejemplo, podría agregar bloques try-except alrededor de la solicitud de la API e incluir diferentes mensajes de respuesta según el tipo de error que ocurra.
Con estos cambios, el código de la API podría verse así:
```
from fastapi import FastAPI
from pydantic import BaseModel
from metrics import tweet_id, create_url, Summer, clear
from typing_extensions import Self

app = FastAPI()

class Username(BaseModel):
    username:str       

@app.post('/points')
def get_points(username: Username):
    tweets_list = tweet_id(username.username)
    tweets_list.get_tweet_ids(str(username.username))
    clear()
    main_output = create_url()
    json_response = connect_to_endpoint(main_output)
    append_to_database(json_response)
    points = Summer().sum_fun()
    return {'points': points}

if __name__ == "__main__":
    uvicorn.run("api:app", host="127.0.0.1", port=5000, log_level="info")
```
Tenga en cuenta que hay partes del script que faltan, como la implementación de los métodos connect_to_endpoint y append_to_database. Asegúrese de incluirlos en función de la implementación original.

Comments are closed.

es.davy.ai

¿Tienes una pregunta?

¿Por qué fastAPI no está creando la base de datos y devolviendo el resultado aquí?

Answer

davy.ai