I’m developing an application using FastAPI and I’m using SQLAlchemy to interact with a PostgreSQL database. In my code, I’m encountering the error sqlalchemy.exc.UnboundExecutionError: Could not locate a bind configured on mapper Mapper[CasinoPlayer(casino_players)] or this Session, and I’m not quite sure how to fix it.
Here’s my code
main.py
from fastapi import FastAPI
from sqlalchemy import create_engine, URL
from sqlalchemy.orm import sessionmaker
from models import Base
from typing import List
from schemas import Player, PlayerInfo
from managers import PlayerManager
url = URL.create(
drivername="postgresql",
host="localhost",
port=5432,
database="casino_player",
username="postgres",
password="postgres"
)
engine = create_engine(url)
Base.metadata.create_all(engine)
Session = sessionmaker(bind=engine)
app = FastAPI(title="Casino Players", version="1.0.0")
player_manager = PlayerManager()
@app.post("/add_player/", response_model=Player)
def add_player(name: str, win_value: float, casino_name: str):
player = Player(name=name, win_value=win_value, casino_name=casino_name)
db_player = player_manager.add_player(player)
return Player(id=db_player.id, name=db_player.name, win_value=db_player.win_value, casino_name=db_player.casino_name)
@app.get("/player/{player_id}", response_model=PlayerInfo)
def get_player_info(player_id: int):
db_player = player_manager.get_player(player_id)
win_value_formatted = f"{db_player.win_value} $"
return PlayerInfo(id=db_player.id, name=db_player.name, win_value=win_value_formatted, casino_name=db_player.casino_name)
@app.get("/all_players/", response_model=List[PlayerInfo])
def get_all_players_info():
players_info_list = player_manager.get_all_players_info()
return players_info_list
@app.put("/player/{player_id}")
def update_player(player_id: int, name: str, win_value: float, casino_name: str):
player_manager.update_player(player_id, name, win_value, casino_name)
return {"message": "Інформація про гравця оновлена"}
@app.delete("/player/{player_id}")
def delete_player(player_id: int):
player_manager.delete_player(player_id)
return {"message": "Гравця успішно видалено"}
@app.delete("/all_players/")
def delete_all_players():
num_deleted = player_manager.delete_all_players()
return {"message": f"Всі гравці видалені, всього {num_deleted} записів"}
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="127.0.0.1", port=8000, log_level="info")
managers.py
from typing import List, Type
from fastapi import HTTPException
from sqlalchemy.orm import Session
from models import CasinoPlayer
from schemas import Player, PlayerInfo
class PlayerManager:
def __init__(self):
self.session = Session()
def add_player(self, player: Player) -> CasinoPlayer:
with self.session as session:
db_player = CasinoPlayer(name=player.name, win_value=player.win_value, casino_name=player.casino_name)
session.add(db_player)
session.commit()
return db_player
def get_player(self, player_id: int) -> Type[CasinoPlayer]:
with self.session as session:
try:
return session.query(CasinoPlayer).filter_by(id=player_id).one()
except:
raise HTTPException(status_code=404, detail="Гравця з таким ID не знайдено")
def get_all_players_info(self) -> List[PlayerInfo]:
with self.session as session:
players_info_list = []
players_records = session.query(CasinoPlayer).all()
for player in players_records:
win_value_formatted = f"{player.win_value} $"
players_info_list.append(PlayerInfo(id=player.id, name=player.name, win_value=win_value_formatted, casino_name=player.casino_name))
return players_info_list
def update_player(self, player_id: int, new_name: str, new_win_value: float, new_casino_name: str):
with self.session as session:
db_player = self.get_player(player_id)
db_player.name = new_name
db_player.win_value = new_win_value
db_player.casino_name = new_casino_name
session.commit()
def delete_player(self, player_id: int):
with self.session as session:
db_player = self.get_player(player_id)
session.delete(db_player)
session.commit()
def delete_all_players(self) -> int:
with self.session as session:
players_records = session.query(CasinoPlayer).all()
num_deleted = 0
for player in players_records:
session.delete(player)
num_deleted += 1
session.commit()
return num_deleted
schemas.py
from typing import Optional
from pydantic import BaseModel, Field, PositiveInt, PositiveFloat
class Player(BaseModel):
id: Optional[PositiveInt] = None
name: str = Field(min_length=1, max_length=50, description="Прізвище, Ім'я, По-батькові")
win_value: PositiveFloat
casino_name: str = Field(min_length=1, max_length=50, description="Назва казино")
class PlayerInfo(BaseModel):
id: PositiveInt
name: str
win_value: str
casino_name: str
models.py
from sqlalchemy import Column, Integer, String, Float
from sqlalchemy.orm import declarative_base
Base = declarative_base()
class CasinoPlayer(Base):
__tablename__ = "casino_players"
id = Column(Integer, primary_key=True, autoincrement=True)
name = Column(String, nullable=False)
win_value = Column(Float, nullable=False)
casino_name = Column(String, nullable=False)
I suspect the error is related to an incorrect binding of the session with my database engine or with the CasinoPlayer model. How can I fix this error and properly configure the SQLAlchemy session to work with FastAPI?
Any help or advice would be greatly appreciated!
Thank you!
Anton Anpilohov is a new contributor to this site. Take care in asking for clarification, commenting, and answering.
Check out our Code of Conduct.