How can i get the response from another API to My API FastApi and store it on postgres

[tkngoualadjo]

First Check

• I added a very descriptive title to this issue.
• I used the GitHub search to find a similar issue and didn't find it.
• I searched the FastAPI documentation, with the integrated search.
• I already searched in Google "How to X in FastAPI" and didn't find any information.
• I already read and followed all the tutorial in the docs and didn't find an answer.
• I already checked if it is not related to FastAPI but to Pydantic.
• I already checked if it is not related to FastAPI but to Swagger UI.
• I already checked if it is not related to FastAPI but to ReDoc.

Commit to Help

• I commit to help with one of those options 👆

Example Code

from typing import Optional
from fastapi import Body, FastAPI, Response, status, HTTPException
from pydantic import BaseModel
from random import randrange
import psycopg2
from psycopg2.extras import RealDictCursor
import time

app = FastAPI()

# model
class Post(BaseModel):
    title: str
    content: str
    published: bool = True
    #rating: Optional[int] = None

#Database connection

while True:
    try:
        conn = psycopg2.connect(host= 'localhost', database='fastapi', user='postgres',
                            password='root', cursor_factory=RealDictCursor)
        cursor = conn.cursor()
        print("Database connection was succesfull")
        break
    except Exception as error:
        print("Connecting to database failed")
        print("Error:", error)
        time.sleep(2)




my_posts = [{"title": "title of post 1", "content": "content of post 1","id": 1}, {"title":
 "favorite foods", "content": "I like pizza", "id":2 }]

def find_post(id):
    for p in my_posts:
        if p['id'] == id:
            return p


def find_index_post(id):
    for i, p in enumerate(my_posts):
        if p['id'] == id:
            return i

@app.get("/")
def root():   
    return {"message": "Hello world"}

#retrieve posts 
@app.get("/posts")
def get_post():
    cursor.execute("""SELECT * FROM posts """)
    posts = cursor.fetchall()
    print(posts)
    return{"data": posts}

# put post in database
@app.post("/posts", status_code=status.HTTP_201_CREATED)
def create_posts(post: Post):
    cursor.execute(""" INSERT INTO posts (title, content, published) VALUES (%s, %s, %s) RETURNING * """,   (post.title, post.content, post.published))

    new_post = cursor.fetchone()
    #conn.commit() for save in database 
    conn.commit()

    return {"data": new_post}

#retrieve one post in database 
@app.get("/posts/{id}")
def get_post(id: int, response: Response):
    cursor.execute(""" SELECT * FROM posts WHERE id = %s """, (str(id)))
    post = cursor.fetchone()
    if not post:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND,
                            detail = f"post with id: {id} was not found")
    return{"post_detail": post}

Description

My api is working fine. I would like to retrieve from my API, the answer of another API (GBFS): https://data.lime.bike/api/partners/v2/gbfs/paris/gbfs.json, which brings up the real-time data of bike and scooter information in the city of Paris. Then store in my postgres database.

The response of the api returns several endpoints:
and I’d like to create a table for each endpoint and store the data.

for exemple :
this one : https://data.lime.bike/api/partners/v2/gbfs/paris/station_information

{
"last_updated": 1670426007,
"ttl": 0,
"version": "2.2",
"data": {
"stations": [
{
"station_id": "paris",
"name": "Paris",
"lat": 48.829,
"lon": 2.3898
}
]
}
}

Operating System

macOS

Operating System Details

MacOS Ventura 13.0.1 on Apple Silicon.

FastAPI Version

0.88.0

Python Version

Python 3.9.2

Additional Context

No response

[fcrozetta]

I may be mistaken here, but this does not seem to be related to FastAPI itself, but a general question on python.
It seems that you want your API to consume another api, and save locally for you. In this case, you can create a "service" that will retrieve the data from the third party api and save it in your DB.

You can trigger this function via an endpoint in your API if you want.

[raphaelauv]

@tkngoualadjo could please close the issue , thanks