Skip to content

Instantly share code, notes, and snippets.

@jsmsalt
Last active June 18, 2024 07:07
Show Gist options
  • Save jsmsalt/26bf25844870d59eee17997727e3a631 to your computer and use it in GitHub Desktop.
Save jsmsalt/26bf25844870d59eee17997727e3a631 to your computer and use it in GitHub Desktop.
Seeding data to database using SQLAlchemy and FastAPI
# The simplest solution I found is to set up an event for each table that executes a method after the table is created.
# Database initial data
INITIAL_DATA = {
'users': [
{
'username': 'superuser',
'email': 'superuser@example.com',
'hashed_password': hash_password('123')
},
{
'username': 'admin',
'email': 'admin@example.com',
'hashed_password': hash_password('123')
}
],
'sometable': [
{'column1': 'value', 'column2': 'value'}
]
}
# This method receives a table, a connection and inserts data to that table.
def initialize_table(target, connection, **kw):
tablename = str(target)
if tablename in INITIAL_DATA and len(INITIAL_DATA[tablename]) > 0:
connection.execute(target.insert(), INITIAL_DATA[tablename])
# In main.py
from sqlalchemy import event
from models.users import User
from models.sometable import SomeTable
# I set up this event before table creation
event.listen(User.__table__, 'after_create', initialize_table)
event.listen(SomeTable.__table__, 'after_create', initialize_table)
app = FastAPI()
# ....
# This will create the DB schema and trigger the "after_create" event
@app.on_event("startup")
def configure():
Base.metadata.create_all(bind=engine)
if __name__ == "__main__":
uvicorn.run(app, host="0.0.0.0", port=8000)
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment