Aún no tienes acceso a esta clase

Crea una cuenta y continúa viendo este curso

Publicando Tweets

23/25
Recursos

Aportes 13

Preguntas 2

Ordenar por:

¿Quieres ver más aportes, preguntas y respuestas de la comunidad? Crea una cuenta o inicia sesión.

Los cursos son tan buenos que es obligado un curso avanzado para realizar conexión con bases de datos.

Recuerden escribir [ ] en el archivo tweets.json o les lanzará error

Yo lo que hice para castear de forma automática los campos a str sin tener que hacerlo cada uno fue poner en el dump el parámetro default, el cual automáticamente guarda todos los campos sin importar que tan escondidos estén en su formato str

with open("tweets.json", "r+", encoding="utf-8") as f:
        results = json.load(f)
        tweet_dict = tweet.dict()
        results.append(tweet_dict)
        f.seek(0)
        json.dump(results,f, default=str, indent=4)

lo mismo que con los usuarios

with open("users.json", "r+", encoding="utf-8") as f:
        results = json.load(f)
        user_dict = user.dict()
        results.append(user_dict)
        f.seek(0)
        json.dump(results,f,default=str,indent=4)

main sin errores:

# Python
import json
from uuid import UUID
from datetime import date
from datetime import datetime
from typing import Optional, List

# Pydantic
from pydantic import BaseModel
from pydantic import EmailStr
from pydantic import Field

# FastAPI
from fastapi import FastAPI
from fastapi import status
from fastapi import Body

app = FastAPI()

# Models

class UserBase(BaseModel):
    user_id: UUID = Field(...)
    email: EmailStr = Field(...)

class UserLogin(UserBase):
    password: str = Field(
        ...,
        min_length=8,
        max_length=64
    )

class User(UserBase):
    first_name: str = Field(
        ...,
        min_lenght=1,
        max_length=50
    )
    last_name: str = Field(
        ...,
        min_lenght=1,
        max_length=50
    )
    birth_date: Optional[date] = Field(default=None)

class UserRegister(User):
    password: str = Field(
        ...,
        min_length=8,
        max_length=64
    )

class Tweet(BaseModel):
    tweet_id: UUID = Field(...)
    content:str = Field(
        ...,
        min_length=1,
        max_length=256
    )
    created_at: datetime = Field(default=datetime.now())
    updated_at: Optional[datetime] = Field(default=None)
    by: User = Field(...)

# Path Operations

## Users

### Register a user
@app.post(
    path="/signup",
    response_model=User,
    status_code=status.HTTP_201_CREATED,
    summary="Register a User",
    tags=["Users"]
)
def signup(user: UserRegister = Body(...)):
    """
    Signup

    This path operations register a user in the app

    Parameters:
    - Request body parameter
        - user: UserRegister

    Return a json with the basic user information:
    - user_id: UUID
    - email: Emailstr
    - first_name: str
    - last_name: str
    - birth_date: datetime
    """
    with open("users.json", "r+", encoding="utf-8") as f:
        results = json.loads(f.read())
        user_dict = user.dict()
        user_dict["user_id"] = str(user_dict["user_id"])
        user_dict["birth_date"] = str(user_dict["birth_date"])
        results.append(user_dict)
        f.seek(0)
        f.write(json.dumps(results))
        return user


### Login a user
@app.post(
    path="/login",
    response_model=User,
    status_code=status.HTTP_200_OK,
    summary="Login a User",
    tags=["Users"]
)
def login():
    pass

### Show all user
@app.get(
    path="/users",
    response_model=List[User],
    status_code=status.HTTP_200_OK,
    summary="Show all users",
    tags=["Users"]
)
def show_all_users():
    """
    This path operation shows all users in the app

    Parameters:
    -

    Returns a json list with all users in the app, with the following keys:
    - user_id: UUID
    - email: Emailstr
    - first_name: str
    - last_name: str
    - birth_date: datetime
    """
    with open("users.json", "r", encoding="utf-8") as f:
        results = json.loads(f.read())
        return results

### Show a user
@app.get(
    path="/users/{user_id}",
    response_model=User,
    status_code=status.HTTP_200_OK,
    summary="show a User",
    tags=["Users"]
)
def show_a_user():
    pass

### Delete a user
@app.delete(
    path="/users/{user_id}/delete",
    response_model=User,
    status_code=status.HTTP_200_OK,
    summary="Delete a User",
    tags=["Users"]
)
def delete_a_user():
    pass

### Update a user
@app.put(
    path="/users/{user_id}/update",
    response_model=User,
    status_code=status.HTTP_200_OK,
    summary="Update a User",
    tags=["Users"]
)
def update_a_user():
    pass

## Tweets

### Show all tweets
@app.get(
    path="/",
    response_model=List[Tweet],
    status_code=status.HTTP_200_OK,
    summary="Show all tweets",
    tags=["Tweets"]
)
def home():
    return {"Twitter API": "Working!"}

### Post a tweet
@app.post(
    path="/post",
    response_model=Tweet,
    status_code=status.HTTP_201_CREATED,
    summary="Post a tweet",
    tags=["Tweets"]
)
def post(tweet: Tweet = Body(...)):
    """
    Post a Tweet

    This path operations post a tweet in the app

    Parameters:
    - Request body parameter
        - tweet: Tweet

    Return a json with the basic tweet information:
    - tweet_id: UUID
    - content:str
    - created_at: datetime
    - updated_at: Optional[datetime]
    - by: User
    """
    with open("tweets.json", "r+", encoding="utf-8") as f:
        results = json.loads(f.read())
        tweet_dict = tweet.dict()
        tweet_dict["tweet_id"] = str(tweet_dict["tweet_id"])
        tweet_dict["created_at"] = str(tweet_dict["created_at"])
        if tweet_dict["updated_at"] is not None:
            tweet_dict["updated_at"] = str(tweet_dict["updated_at"])
        tweet_dict["by"]["user_id"] = str(tweet_dict["by"]["user_id"])
        tweet_dict["by"]["birth_date"] = str(tweet_dict["by"]["birth_date"])
        results.append(tweet_dict)
        f.seek(0)
        f.write(json.dumps(results))
        return tweet

### Show a tweet
@app.get(
    path="/tweets/{tweet_id}",
    response_model=Tweet,
    status_code=status.HTTP_200_OK,
    summary="Show a tweet",
    tags=["Tweets"]
)
def show_a_tweet():
    pass

### Delete a tweet
@app.delete(
    path="/tweets/{tweet_id}/delete",
    response_model=Tweet,
    status_code=status.HTTP_200_OK,
    summary="Delete a tweet",
    tags=["Tweets"]
)
def delete_a_tweet():
    pass

### Update a tweet
@app.put(
    path="/tweets/{tweet_id}/update",
    response_model=Tweet,
    status_code=status.HTTP_200_OK,
    summary="Update a tweet",
    tags=["Tweets"]
)
def update_a_tweet():
    pass

TypeError: Object of type UUID is not JSON serializable

Si les aparece ese error, sigan con la clase que Facundo se los explica, y no pierdan tiempo como yo. Saludos

Como la fecha de cumpleaños no es un dato que debería recuperar con cada tweet, yo cambié el campo by de User a UserBase, así sólo se debe castear el user_id:

class Trino(BaseModel):
    id: UUID = Field(
        ...,
        title="Trino Id"
        )
    content: str = Field(
        ...,
        title="Trino content",
        min_length=1,
        max_length=256
        )
    created_at: datetime = Field(
        default=datetime.now(),
    )
    updated_at: Optional[datetime] = Field(
        default=datetime.now(),
    )
    by: UserBase = Field(
        ...,
        title="User who created the trino"
        )

@router.post(
    path="/trino",
    response_model=Trino,
    status_code=status.HTTP_201_CREATED,
    summary="Post a trino",
    tags=["Trino"]
    )
def post_trino(trino: Trino):
    """
    Post a trino

    This endpoint post a trino.

    Parameters:
    trino: Trino object

    Returns:
    Dictionary with the trino.
    """
    with open("data/trinos.json", "r+", encoding="utf-8") as f:
        trinos = json.load(f)
        trino_dict = trino.dict()
        trino_dict["id"] = str(trino_dict["id"])
        trino_dict["created_at"] = str(trino_dict["created_at"])
        trino_dict["updated_at"] = str(trino_dict["updated_at"])
        trino_dict["by"]["id"] = str(trino_dict["by"]["id"])
        trinos.append(trino_dict)
        f.seek(0)
        json.dump(trinos, f, indent=4)
        f.truncate()
    return trino

Algunas cosas como el created date, tweet id y ese tipo de cosas no es como que el usuario las ponga el mismo, entonces decidí en la misma logica de la funcion que el mismo backend de la app sea el encargado de setear un random UUID al tweet y de setear su fecha de creacion (la fecha y hora UTC del momento en el que se hizo el http request). Me quedó así:

def post_a_tweet(tweet: TweetIn = Body(...)):
    with open('./data/tweets.json', 'r+', encoding='utf-8') as f:
        list_of_tweets = json.load(f)
        tweet = dict(tweet)

        #Casting user atributes
        tweet['user'] = dict(tweet['user']) 
        tweet['user']['user_id'] = tweet['user']['user_id'].hex
        tweet['user']['birth_date'] = tweet['user']['birth_date'].isoformat()
        if tweet['user']['gender'] is not None:
            tweet['user']['gender'] = str(tweet['user']['gender']).rpartition('.')[2]

        #Adding some tweet atributes
        tweet['tweet_id'] = uuid.uuid1().hex
        tweet['created_at'] = datetime.utcnow().replace(tzinfo=tz.tzutc()).isoformat() 
        tweet['updated_at'] = None
        list_of_tweets.append(tweet)
        f.seek(0)
        json.dump(list_of_tweets, f)

    return tweet

docstring markdown

  '''
    Post a Tweet

    This path operation post a Tweet in the app
    
    Parameters:
    - Request Body parameter
        - tweet: Tweet
    
    Returns a JSON with the basic tweet information:
    - tweet_id: UUID
    - content: str
    - created_at: datetime
    - updated_at: Optional[datetime]
    - by: User
    '''

En el docstring se puede escribir con lo sintaxsis Markdown solo deben tener en cuenta la identación.
https://en.wikipedia.org/wiki/Markdown

Primer borrador de la funcion

def file_work(file_name:str,mod:str,objeto):
    with open(file_name,mod,encoding="utf-8")as f:
        results = json.loads(f.read())
        new_dict= objeto.dict()
        results.append(new_dict)
        f.seek(0)
        json.dump(results,f,default=str,indent=4)

se podria agregar como parametro la operacion a realizar y ejecutar el comando CRUD segun corresponda

Lo Bueno de Adicionar cosas Que el profesor no da en la clase mientras la clase va poco a poco vas practicando lo de cursos anteriores, Lo Malo es que Te Saltan errores que te pones la mano en la cabeza Pero Realmente lo Bueno de eso malo Es que te vez obligado a Solucionar que no lo consigues ni en la clase y ni en los comentario Que es cuando sucede la magia y uno aprende mas

Asi me quedó la funcion para publicar un tweet, lo que hace es generar un nuevo uuid y ademas a la llave “by” le da el valor de un usuario ya registrado en nuestra “db” , tambien como response envio el tweet_dict por que en la variable tweet no se ven las modificaciones que se hicieron.

    with open("tweets.json", "r+", encoding="utf-8") as f:
        results = json.load(f)
        tweet_dict = tweet.dict()
        tweet_dict["tweet_id"] = str(uuid4())
        tweet_dict["created_at"] = str(datetime.now())

        with open("users.json", "r", encoding="utf-8") as u:
            list_users = json.load(u)
            user: User = random.choice(list_users)

        tweet_dict["by"] = user
        results.append(tweet_dict)
        f.seek(0)
        json.dump(results, f, default=str, indent=4)
    return tweet_dict

En el minuto 15:40 se vio un momento magico en la clase como casteo la variable user_id dentro del json by, ese tipo de errores a mi me tomarian horas pero de estas clases se aprende realmente.