recently I have been trying to educate myself on REST APIs.
As a part of that, I have created a small project using FastAPI in python and I would love to hear some feedback about it.
When I wrote this, I didn't take into consideration any design pattern, and honestly, I almost knew nothing about them - on the last week I have been trying to educate myself on that subject as well, and I would appreciate some advice on what design pattern should I use for this API? ( I think Pagination is quite fitting ).
The code purpose is -
- query database for pokemons information by parameters such as name and type
- Create a new pokemon (add to the database)
- Delete Pokemon from the database based on name
- Update pokemon information based on name
This project relies on two main files:
pokeapi.py:
from typing import Optional
from fastapi import FastAPI, Path, HTTPException, status
from pydantic import BaseModel
from database import get_poke_by_name, get_poke_by_type, add_poke_to_db, \
update_poke, delete_poke
app = FastAPI()
class Pokemon(BaseModel):
name: str
primary_type: str
secondary_type: str
sum_stats: int
hit_points: int
attack_strength: int
defensive_strength: int
special_attack_strength: int
special_defensive_strength: int
@app.get("/")
def root():
raise HTTPException(status_code=status.HTTP_200_OK,
detail="Welcome to PokeAPI")
@app.get("/poke/{pokemon_name}")
def get_pokemon_by_name(pokemon_name: str = Path(None,
description="Name of the "
"pokemon you'd "
"like to "
"retrieve")):
pokemon = get_poke_by_name(pokemon_name)
if not pokemon:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND,
detail="Pokemon not found")
return {"Pokemon": pokemon[0],
"Types": [pokemon[1], pokemon[2]],
"HP": pokemon[4],
"Attack": pokemon[5],
"Special Attack": pokemon[6],
"Defense": pokemon[7],
"Special Defense": pokemon[8],
}
@app.get("/poketype/{poke_type}")
def get_pokemon_by_type(poke_type: str = Path(None,
description="Primary type of "
"the pokemons you "
"want to query"),
type2: Optional[str] = None):
pokemons = get_poke_by_type(poke_type, type2)
if not pokemons:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND,
detail="No pokemon with this type")
result = {}
for idx, pokemon in enumerate(pokemons):
result[idx] = {"Pokemon": pokemon[0],
"Types": [pokemon[1], pokemon[2]],
"HP": pokemon[4],
"Attack": pokemon[5],
"Special Attack": pokemon[6],
"Defense": pokemon[7],
"Special Defense": pokemon[8],
}
return result
@app.post("/newPoke/{pokemon_name}")
def create_pokemon(pokemon_name: str, pokemon: Pokemon):
if get_poke_by_name(pokemon_name):
raise HTTPException(status_code=status.HTTP_406_NOT_ACCEPTABLE,
detail="Pokemon already exists")
add_poke_to_db(pokemon.name, pokemon.primary_type, pokemon.secondary_type,
pokemon.sum_stats, pokemon.hit_points,
pokemon.attack_strength, pokemon.special_attack_strength,
pokemon.defensive_strength,
pokemon.special_defensive_strength)
raise HTTPException(status_code=status.HTTP_201_CREATED,
detail="Pokemon created successfully")
@app.put("/updatePoke/{pokemon_name}")
def update_pokemon(pokemon_name: str, pokemon: Pokemon):
if not get_poke_by_name(pokemon_name):
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND,
detail="Pokemon not found")
update_poke(pokemon.name, pokemon.primary_type, pokemon.secondary_type,
pokemon.sum_stats, pokemon.hit_points,
pokemon.attack_strength, pokemon.special_attack_strength,
pokemon.defensive_strength,
pokemon.special_defensive_strength)
raise HTTPException(status_code=status.HTTP_200_OK,
detail="Pokemon details updated")
@app.delete("/deletePoke/{pokemon_name}")
def delete_pokemon(pokemon_name: str):
if not get_poke_by_name(pokemon_name):
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND,
detail="Pokemon not found")
delete_poke(pokemon_name)
raise HTTPException(status_code=status.HTTP_200_OK,
detail="Pokemon deleted successfully")
and database.py
from pathlib import Path
import sqlite3
import pandas as pd
DB_FILENAME = "poke_db.db"
def table_exists(cursor):
cursor.execute('''
SELECT count(name) FROM sqlite_master WHERE type='table' AND name='Pokemons' ''')
if not cursor.fetchone()[0]:
return False
return True
def init_db():
if not Path(DB_FILENAME).is_file():
Path(DB_FILENAME).touch()
def load_csv_to_db():
init_db()
conn = sqlite3.connect(DB_FILENAME)
cursor = conn.cursor()
cursor.execute('''
CREATE TABLE IF NOT EXISTS Pokemons (name text, type1 text,
type2 text, sum_stats int, hp int, attack int,
special_attack int, defense int, special_defense int)
''')
poke_data = pd.read_csv('Pokemon.csv')
poke_data.drop(['#', 'Speed', 'Generation', 'Legendary'], axis=1, inplace=True)
poke_data.columns = ['name', 'type1', 'type2', 'sum_stats',
'hp', 'attack', 'special_attack', 'defense',
'special_defense']
poke_data.to_sql('Pokemons', conn, if_exists='append', index=False)
def get_poke_by_name(poke_name):
conn = sqlite3.connect(DB_FILENAME)
cursor = conn.cursor()
if not table_exists(cursor):
load_csv_to_db()
cursor.execute('''SELECT * FROM Pokemons WHERE name = ?''', (poke_name,))
return cursor.fetchone()
def get_poke_by_type(type1, type2):
conn = sqlite3.connect(DB_FILENAME)
cursor = conn.cursor()
if not table_exists(cursor):
load_csv_to_db()
if type2:
cursor.execute('''
SELECT * FROM Pokemons WHERE type1 = ? AND type2 = ?''', (type1, type2))
else:
cursor.execute('''
SELECT * FROM Pokemons WHERE type1 = ?''', (type1,))
return cursor.fetchall()
def add_poke_to_db(name, type1, type2, sum_stats, hp, attack, special_attack,
defense, special_defense):
conn = sqlite3.connect(DB_FILENAME)
cursor = conn.cursor()
if not table_exists(cursor):
load_csv_to_db()
cursor.execute('''
INSERT INTO Pokemons ('name', 'type1', 'type2', 'sum_stats',
'hp', 'attack', 'special_attack', 'defense', 'special_defense')
VALUES (?,?,?,?,?,?,?,?,?)''', (name, type1, type2, sum_stats, hp, attack,
special_attack, defense, special_defense))
conn.commit()
def update_poke(name, type1=None, type2=None, sum_stats=None, hp=None,
attack=None, special_attack=None, defense=None,
special_defense=None):
conn = sqlite3.connect(DB_FILENAME)
cursor = conn.cursor()
if not table_exists(cursor):
load_csv_to_db()
params = [type1, type2, sum_stats, hp, attack, special_attack,
defense, special_defense]
params_names = ['type1', 'type2', 'sum_stats', 'hp', 'attack',
'special_attack', 'defense', 'special_defense']
for param, param_name in zip(params, params_names):
if param:
query = '''
UPDATE Pokemons SET ''' + param_name + '''
= ? WHERE name = ?'''
cursor.execute(query, (param, name))
conn.commit()
def delete_poke(name):
conn = sqlite3.connect(DB_FILENAME)
cursor = conn.cursor()
if not table_exists(cursor):
load_csv_to_db()
cursor.execute('''
DELETE FROM Pokemons WHERE name = ?''', (name,))
conn.commit()
If it will be more comfortable, here's a link to the GitHub repo https://github.com/DevEliran/PokeAPI
I'd love your review on any small or big detail
