testdrivenio / fastapi-sqlmodel-alembic Goto Github PK
View Code? Open in Web Editor NEWSample FastAPI project that uses async SQLAlchemy, SQLModel, Postgres, Alembic, and Docker.
License: MIT License
Sample FastAPI project that uses async SQLAlchemy, SQLModel, Postgres, Alembic, and Docker.
License: MIT License
First, thanks for this great tutorial.
With this async approach, how is it possible to get filtered data from the database (using sqlalchemy and sqlmodel)?
SQLModel's documentations suggest using where
, but it doesn't work for async approach and they don't have any documentation for this yet.
I tried the following, but didn't work:
@app.get("/songs/{id}", response_model=list[Song])
def get_song(id: int = Path(..., gt=0), session: Session = Depends(get_session)):
result = session.execute(select(Song).where(Song.id == id))
songs = result.scalars().all()
return [Song(name=song.name, artist=song.artist, id=song.id) for song in songs]
Either we use SQLAlchemy. Or we use completely SQLModel. Sqlmodel is built on top SQLAlchemy and Pydantic
I followed this post , everything is fine until I add "year" item in "SongBase", and when I execute
python -m alembic revision --autogenerate -m "add year in SongBase"
the autogenerated migration file is not the same as yours, it's like
def upgrade():
# ### commands auto generated by Alembic - please adjust! ###
op.add_column('song', sa.Column('year', sa.Integer(), nullable=True))
op.alter_column('song', 'id',
existing_type=sa.INTEGER(),
nullable=True,
autoincrement=True)
op.create_index(op.f('ix_song_year'), 'song', ['year'], unique=False)
# ### end Alembic commands ###
def downgrade():
# ### commands auto generated by Alembic - please adjust! ###
op.drop_index(op.f('ix_song_year'), table_name='song')
op.alter_column('song', 'id',
existing_type=sa.INTEGER(),
nullable=False,
autoincrement=True)
op.drop_column('song', 'year')
# ### end Alembic commands ###
and of course fail to execute "python -m alembic upgrade head", because db id cannot be null.
Even if I didn't change anything and execute autogenerate again, the migration file still have
op.alter_column('song', 'id',
existing_type=sa.INTEGER(),
nullable=True,
autoincrement=True)
Could you help me find out why does it happen?
Environment:
Ubuntu 20.04 wsl2
Python 3.9.7
models.py
from typing import Optional
from sqlmodel import Field, SQLModel
class SongBase(SQLModel):
name: str
artist: str
year: Optional[int] = None
class Song(SongBase, table=True):
id: int = Field(default=None, primary_key=True)
class SongCreate(SongBase):
pass
The curl command uses "year":"2021". But year is defined as an int It should be "year":2021.
I don't know how to repair this but I didn't find any way to do "alembic upgrade head" in docker-compose. But the fact that it's useless is really disgusting. Usually people use docker for simple work on servers and it's main idea to unite a big amount of commads instead of exexcuting it by hands. So, you didn't find how to fix it and just said "do a docker-compose exec web alembic upgrade head every time you need to up docker-compose". I hate myself that i didn't find how to fix it but more I hate for this subquality work
I wonder if
@app.get("/songs", response_model=list[Song])
async def get_songs(session: AsyncSession = Depends(get_session)):
result = await session.execute(select(Song))
return result.scalars().all()
wouldn't be the same but shorter then
@app.get("/songs", response_model=list[Song])
async def get_songs(session: AsyncSession = Depends(get_session)):
result = await session.execute(select(Song))
songs = result.scalars().all()
return [Song(name=song.name, artist=song.artist, year=song.year, id=song.id) for song in songs]
or is it necessary to create the new models for some reason?
What is the best method for inserting default data?
Sqlalchemy's event listeners for "on_create" seems like overkill.
Without using Alembic, is there some other SQLmodel'ish way to create initial data?
Love SQLModel! Got confused in the start because I Pulled HEAD instead of base for the tutorial start.
Thanks Mike, appreciate the help this repo provided.
One question though: Where are the tests? Isn't this test-driven.io ๐
I read another article you posted (https://testdriven.io/blog/fastapi-crud/) and noticed you use monkey patching to make the tests pass. Is there no way to use pytest + async db/routes without monkeypatching?
firstly: thanks so much for this - it allowed me to get started with relational databases in fastapi really quickly. Thanks!!
I noticed a couple of things:
it looks like alembic really wants you to set nullable=False
fo rthe id field - if not every subsequent migration will try and do something weird to the id
column. This confused me for a lot longer than I'll be willing to admit.
instead of keeping two sets of db settings (one for alembic and one for the app) it might be better to use a shared BaseSettings
object to keep track of that:
settings.py
:
from pydantic import BaseSettings
class Settings(BaseSettings):
db_name: str = "app"
db_user: str = "postgres"
db_password: str = "postgres"
db_host: str = "localhost"
db_port: int = 5432
def db_conn_string(self):
return f"postgresql+asyncpg://{self.db_user}:{self.db_password}@{self.db_host}:{self.db_port}/{self.db_name}"
in db.py
:
settings = Settings()
engine = create_async_engine(settings.db_conn_string(), echo=True, future=True)
in env.py
:
import asyncio
from logging.config import fileConfig
from alembic import context
from sqlalchemy.ext.asyncio import create_async_engine
from sqlmodel import SQLModel
from app.settings import Settings
settings = Settings()
config = context.config
fileConfig(config.config_file_name)
target_metadata = SQLModel.metadata
def run_migrations_offline():
url = settings.db_conn_string()
context.configure(
url=url,
target_metadata=target_metadata,
literal_binds=True,
dialect_opts={"paramstyle": "named"},
)
with context.begin_transaction():
context.run_migrations()
def do_run_migrations(connection):
context.configure(connection=connection, target_metadata=target_metadata)
with context.begin_transaction():
context.run_migrations()
async def run_migrations_online():
connectable = create_async_engine(settings.db_conn_string(), echo=True, future=True)
async with connectable.connect() as connection:
await connection.run_sync(do_run_migrations)
if context.is_offline_mode():
run_migrations_offline()
else:
asyncio.run(run_migrations_online())
i don't know if you take PRs for this project, but for the first issue I saw some comment in the alembic github discussion that looked very similar to this code so it would probably be a big help to the community if that was fixed. If you want these suggestions in a PR instead I'd be happy to make one.
A declarative, efficient, and flexible JavaScript library for building user interfaces.
๐ Vue.js is a progressive, incrementally-adoptable JavaScript framework for building UI on the web.
TypeScript is a superset of JavaScript that compiles to clean JavaScript output.
An Open Source Machine Learning Framework for Everyone
The Web framework for perfectionists with deadlines.
A PHP framework for web artisans
Bring data to life with SVG, Canvas and HTML. ๐๐๐
JavaScript (JS) is a lightweight interpreted programming language with first-class functions.
Some thing interesting about web. New door for the world.
A server is a program made to process requests and deliver data to clients.
Machine learning is a way of modeling and interpreting data that allows a piece of software to respond intelligently.
Some thing interesting about visualization, use data art
Some thing interesting about game, make everyone happy.
We are working to build community through open source technology. NB: members must have two-factor auth.
Open source projects and samples from Microsoft.
Google โค๏ธ Open Source for everyone.
Alibaba Open Source for everyone
Data-Driven Documents codes.
China tencent open source team.