The code in this example generates GET and POST routes for each of your database tables that returns all the routes.
Note that this is not sufficient to create a REST API for coursework 2. You will be expected to include Pydantic schemas and include PUT and DELETE routes.
-
Include the dependencies in your pyproject.toml:
dependencies = [ "fastapi", "uvicorn", # Plus all your existing dependencies ]
-
Copy the api package to your project structure, e.g. in 'src'. You don't need
paralympics.dbunless you want to test it using that database. -
Replace the database name and path found on line 161 with the location of your database Line 161:
database_location = Path(__file__).parent.joinpath("paralympics.db") -
If you want to generate routes that allow you to search using query parameters, uncomment line 256
app.get(f"/{_t}/search", name=f"{_t}_search")(_make_search_route(_t)). See the docstring for the_make_search_routemethod for more info on how to the use a search route. -
If you need a route that returns data joined from two or more tables, add a custom route and a method to the Data class. An example for the Paralympics database:
- Add a method to the
Dataclass to get the data you want from the database. For example:
def get_all_data(self): """ Method to return all data from the paralympics .db file. Returns: data: json format data Raises: e: Exception """ sql = ( "SELECT country.country_name, games.event_type, games.year, games.start_date, " "games.end_date, host.place_name, games.events, games.sports, games.countries, " "games.participants_m, games.participants_f, games.participants, host.latitude, " "host.longitude " "FROM games " "JOIN games_host ON games.id = games_host.games_id " "JOIN host ON games_host.host_id = host.id " "JOIN country ON host.country_id = country.id" ) try: conn = sqlite3.connect(self.database_file) with conn: conn.row_factory = sqlite3.Row # Returns columns by names instead of tuples cur = conn.cursor() cur.execute(sql) rows = cur.fetchall() if not rows: return [] data = [dict(row) for row in rows] return data except Exception as e: raise RuntimeError(f"Error querying tables: {e}") from e finally: if conn: conn.close()
- Add a custom route in the API code section. For example:
@app.get("/all") async def get_all(): try: return data.get_all_data() except AttributeError: raise HTTPException(status_code=500, detail="ParalympicsData.get_json not implemented") except Exception as exc: raise HTTPException(status_code=500, detail=str(exc))
- Add a method to the
-
Run the app. By default, the app will run on http://127.0.0.1:8000. Go to the route to see the docs which list all the routes that have been created.