Dynamic queries

In the previous section, we made use of SQL statements to interact with a database. However, the SQL statements we wrote were not dynamic yet. We programmed our SQL statement, but no external factor could change anything to our SQL statement.

In reality, we often include dynamic information in our SQL statement. This dynamic information might be information coming along with an API request as query parameter for example. In this section, we will look at how we can incorporate this into our database interaction.

In the example that follows, we want to filter the festivals based on the province in which they take place, so the result we get back from the database should only contain festivals that take place in the province that was passed to our API as query parameter. But first things first. Before we include this in our database interaction, let's first write a new endpoint in main.py that can handle a query parameter.

SQL query in festival_queries.py

We will start by adding an SQL query to festival_queries.py as shown below. Note that %s was used to indicate where our dynamic information will be inserted in the query later on.

festival_name_query = "SELECT name FROM festivaldb.festival;"

festival_name_and_province_query = "SELECT name, province FROM festivaldb.festival;"

festivals_by_province_query = "SELECT name, province FROM festivaldb.festival WHERE province = %s;"

Query parameters in main.py

In the example below, we added a new endpoint to the main.py file that we created in the previous chapter. This endpoint now contains a query parameter, which will be used in an SQL query later on.

from fastapi import FastAPI
import database
from queries import festival_queries as queries

app = FastAPI()

@app.get("/festivals")
def get_all_festivals():
    query = queries.festival_name_query
    festivals = database.execute_sql_query(query)
    if isinstance(festivals, Exception):
        return festivals, 500
    festivals_to_return = []
    for festival in festivals:
        festivals_to_return.append(festival[0])
    return({'festivals': festivals_to_return})

@app.get("/province")
def get_all_festivals_by_province(name: str):
    query = queries.festivals_by_province_query
    festivals = database.execute_sql_query(query, (
        name,
    ))
    if isinstance(festivals, Exception):
        return festivals, 500
    festivals_to_return = []
    for festival in festivals:
        festivals_to_return.append(festival[0])
    return({'festivals': festivals_to_return})

Notice that we created a new path operation decorator and path operation function. The path operation decorator contains the link to our endpoint, which is /province in this case. We also created a new path operation function called get_all_festivals_by_province. This function contains one parameter, name, of the datatype str. As a result, our API now expects a query parameter called name whenever this endpoint is addressed.

In the lines that follow (see below), we retrieve our SQL query from festival_queries.py and use it as parameter of the execute_sql_query() function. This function can be found in the database.py file. It is important to note that a second parameter is passed to the function as well, containing the query parameter name. By passing it like this, the value of name will be inserted in our SQL query before it is executed in our database. Pay attention to how this parameter was added, it is important to use this notation.

query = queries.festivals_by_province_query
festivals = database.execute_sql_query(query, (
    name,
))

SQL query

Let's assume our API was addressed in the following way:

http://127.0.0.1:8000/province?name=Antwerp (opens new window)

Then Antwerp will be the value of the query parameter name. As a result, our SQL query will be executed as follows because %s is now replaced by Antwerp, the value of our query parameter.

SELECT name, province FROM festivaldb.festival WHERE province = 'Antwerp';

We now make use of formatting with %s to dynamically build up our SQL query. You could also build up the SQL query with a simple concatenation of text:

sql_query = "SELECT name, province FROM festivaldb.festival WHERE province = '" + province + "';"

However, concatenating our text like this poses a serious risk here: we do not know in advance what the string province will contain. It might as well contain the following text: Antwerp'; DROP TABLE festivaldb.festival; --. If this would be concatenated with our SQL query, not only will our query be executed, but also the DROP TABLE query. This is a serious risk called SQL injection.

SQL injection

SQL injection is a type of security vulnerability that occurs when an attacker injects malicious SQL code into input fields of a web application, which might be sent to an API, exploiting vulnerabilities in the application's code. This injected SQL code can manipulate the database queries, allowing the attacker to gain unauthorized access to the database, retrieve or modify sensitive data, execute administrative operations, or even compromise the entire system. It's a significant threat to web applications that interact with databases and can lead to severe consequences if not properly mitigated.

Want to know more?

• Some basic SQL injection information: SQL Injection - W3Schools (opens new window)
• Preventing SQL Injection Attacks With Python: Real Python (opens new window)
• SQL Injection: OWASP (opens new window)

We can prevent SQL injection from happening by using formatting in Python as shown above.

A more complex example

Below you can find an addition to the example that was built up to now. In the example below, not only the festival name is returned in a list, but more information of each festival is given using a dictionary for each festival. In addition, the endpoint requires two query parameters: one for a string that holds (a part of) the name of a festival or festivals and another query parameter that holds an integer indicating the month in which the festival took place (so where either the start date or the end date of the festival is in that month).

In the example below, the added lines of code are highlighted.

main.py

from fastapi import FastAPI
import database
from queries import festival_queries as queries

app = FastAPI()

@app.get("/festivals")
def get_all_festivals():
    query = queries.festival_name_query
    festivals = database.execute_sql_query(query)
    if isinstance(festivals, Exception):
        return festivals, 500
    festivals_to_return = []
    for festival in festivals:
        festivals_to_return.append(festival[0])
    return({'festivals': festivals_to_return})

@app.get("/province")
def get_all_festivals_by_province(name: str):
    query = queries.festivals_by_province_query
    festivals = database.execute_sql_query(query, (
        name,
    ))
    if isinstance(festivals, Exception):
        return festivals, 500
    festivals_to_return = []
    for festival in festivals:
        festivals_to_return.append(festival[0])
    return({'festivals': festivals_to_return})

@app.get("/festivals/name_and_month")
def get_all_festivals_by_name_and_month(name: str, month: int):
    query = queries.festivals_by_name_and_month_query
    festivals = database.execute_sql_query(query, (
        '%{}%'.format(name),
        month,
        month,
    ))
    if isinstance(festivals, Exception):
        return festivals, 500
    festivals_to_return = []
    for festival in festivals:
        location = festival[1] + ' (' + festival[4] + ')'
        festival_dictionary = {"name": festival[0],
                               "startDate": festival[2],
                               "endDate": festival[3],
                               "location": location }
        festivals_to_return.append(festival_dictionary)
    return({'festivals': festivals_to_return})

festival_queries.py

festival_name_query = "SELECT name, province FROM festivaldb.festival;"

festival_name_and_province_query = "SELECT name, province FROM festivaldb.festival;"

festivals_by_province_query = "SELECT name, province FROM festivaldb.festival WHERE province = %s;"

festivals_by_name_and_month_query = "SELECT name, location, startDate, endDate, province FROM festivaldb.festival WHERE name LIKE %s AND (date_format(startDate, '%m') = %s OR date_format(endDate, '%m') = %s);"

Try it out!

Use Postman to send a request to this endpoint and pass the necessary query parameters in the URL. In the example below, we looked for all festivals that have the word 'festival' in their name and took place in August.

We used the following URL to address our endpoint:

http://127.0.0.1:8000/festivals/name_and_month?name=festival&month=8 (opens new window)

In Postman, the result should look as follows: