> ## Documentation Index
> Fetch the complete documentation index at: https://notes.kodekloud.com/llms.txt
> Use this file to discover all available pages before exploring further.
# Fetch Post By Id Sql
> This article explains how to retrieve a post by ID from a PostgreSQL database using FastAPI with secure SQL queries.
In this article, we demonstrate how to retrieve an individual post from a PostgreSQL database using FastAPI. We will build a secure GET endpoint that fetches a post by its ID using a parameterized SQL query, significantly reducing the risk of SQL injection attacks. Additionally, we briefly cover the DELETE endpoint for completeness.
Below is the step-by-step evolution of the implementation along with key points and code examples.
***
## Step 1: Basic GET and DELETE Endpoints
Initially, you define the endpoints for retrieving and deleting posts. Starting with a basic structure, the endpoints might look like this:
```python theme={null}
@app.get("/posts/{id}")
def get_post(id: int):
post = find_post(id)
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}
@app.delete("/posts/{id}", status_code=status.HTTP_204_NO_CONTENT)
def delete_post(id: int):
# Endpoint to delete a post (implementation details omitted)
```
When executing the application, you may see console output similar to the following:
```text theme={null}
INFO: Started server process [22040]
INFO: Waiting for application startup.
INFO: Application startup complete.
INFO: Uvicorn running on http://127.0.0.1:62206 (Press CTRL+C to quit)
INFO: 127.0.0.1:62206 - "POST /posts HTTP/1.1" 201 Created
```
***
## Step 2: Incorporating SQL Queries
To fetch posts from the database, the cursor object is used to execute SQL queries. An initial approach might involve fetching all posts:
```python theme={null}
@app.get("/posts/{id}")
def get_post(id: int):
cursor.execute("SELECT * FROM posts")
post = find_post(id)
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}
@app.delete("/posts/{id}", status_code=status.HTTP_204_NO_CONTENT)
def delete_post(id: int):
# deletion logic here
```
This approach retrieves all records from the posts table. Later, we will optimize the query to ensure that only the specified post is selected.
***
## Step 3: Filtering Posts by the Given ID
Instead of retrieving all posts, we update the SQL statement to fetch a single record that matches the provided ID. For testing purposes, you might initially hardcode the query with a known ID (for example, 1):
```python theme={null}
@app.get("/posts/{id}")
def get_post(id: int):
cursor.execute("SELECT * FROM posts WHERE id = 1")
cursor.fetchone()
post = find_post(id)
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}
@app.delete("/posts/{id}", status_code=status.HTTP_204_NO_CONTENT)
def delete_post(id: int):
```
After verifying the query using a direct SQL command like `SELECT * FROM posts;` in your database client, you can proceed to parameterize the query.
***
## Step 4: Parameterizing the SQL Query
To safeguard against SQL injection, replace the hardcoded ID with the provided path parameter through parameter substitution. Although the path parameter is already validated as an integer, it’s converted to a string for compatibility with the SQL driver. The `%s` placeholder is used in the SQL statement:
```python theme={null}
@app.get("/posts/{id}")
def get_post(id: int):
cursor.execute("SELECT * FROM posts WHERE id = %s", (str(id),))
test_post = cursor.fetchone()
print(test_post)
post = find_post(id)
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}
```
Be sure to include an extra comma in the tuple (i.e., `(str(id),)`) to prevent unexpected issues with parameter tuple assignment.
***
## Step 5: Final Cleaned-Up Code
After removing debugging statements and ensuring robust input validation, the final version of the GET endpoint is as follows. The DELETE endpoint is also updated for completeness:
```python theme={null}
@app.get("/posts/{id}")
def get_post(id: int):
# Execute a parameterized SQL query to safely fetch the post by its ID.
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}
@app.delete("/posts/{id}", status_code=status.HTTP_204_NO_CONTENT)
def delete_post(id: int):
# Find the index of the post in the internal list and remove it.
index = find_index_post(id)
if index is not None:
my_posts.pop(index)
```
When the GET endpoint is called for an existing post (for example, with an ID of 1), the JSON response may appear as:
```json theme={null}
{
"post_detail": {
"id": 1,
"title": "first post",
"content": "some interesting stuff.",
"published": true,
"created_at": "2021-08-21T23:08:39.631328-04:00"
}
}
```
In contrast, when a post is not found (for example, an ID of 5), the API returns an error response:
```json theme={null}
{
"detail": "post with id: 5 was not found"
}
```
***
## Additional Notes
During development, you might encounter an error such as:
```text theme={null}
return await loop.run_in_executor(None, func, *args)
File "C:\Users\sanje\AppData\Local\Programs\Python\Python39\lib\concurrent\futures\thread.py", line 52, in run
```
This error should resolve once the SQL query correctly uses parameter substitution. If issues persist with the parameter tuple, double-check for the extra comma, ensuring it is written as `(str(id),)`.
Happy coding!