Hey, I would like have an example to look up to and also your opinion on using Sanic with SQLite.
Should I completely avoid it and use PostgreSQL if there are plans for large scale use?
This is very much a decision that you will have to answer. What do you mean by “large scale” use? How many instances of Sanic do you intend to run? SQLite is generally more intended for simpler use cases, but this is very much an architectural question that is somewhat outside the scope here.
Sanic will have no problem working with either. But, if you intend to scale with multiple instances, and serve high traffic, then Postgres sounds more the right direction.
Here is far from perfect and unrevised example that seems to work.
I encourage everyone to make a better example and check for missing things, mistakes and errors, as there might be some.
Commands:
pip install -r requirements.txtpython create_sqlite_tables.pysanic server- Use python to test.
1. Requirements.txt
# requirements.txt - pip packages for the project to work.
#
# Usage:
# python install_requirements.py
#
# Alternate
# pip install -r requirements.txt
# A reliable, high-performance Python web framework.
sanic==23.3.0 # Stable version
sanic_cors==2.2.0 # Stable version
#sanic-ext
# Sanic Official SQLite Support
# changing the driver from PostgreSQL aiomysql to asyncpg/aiosqlite.
sqlalchemy
asyncpg
aiosqlite
# Not Official SQLite support
# No support for Windows and no example to use with Sanic.
#PostgreSQLite # Python module that gives you the power of a PostgreSQL server, with the convenience of the sqlite3 module.
2. create_sqlite_tables.py
from sqlalchemy import create_engine, Column, Integer, String, ForeignKey
from sqlalchemy.orm import declarative_base
# Create an engine to connect to the database
engine = create_engine("sqlite:///test.db", echo=True)
# Create a base class for the table models
Base = declarative_base()
# Define the table models
class Person(Base):
__tablename__ = "person"
id = Column(Integer, primary_key=True)
name = Column(String)
class Car(Base):
__tablename__ = "car"
id = Column(Integer, primary_key=True)
brand = Column(String)
user_id = Column(Integer, ForeignKey("person.id"))
# Bind the engine to the Base
Base.metadata.bind = engine
# Create the tables
Base.metadata.create_all(bind=engine)
# Dispose the engine
engine.dispose()
3. server.py
from sanic import Sanic
from sqlalchemy import select
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import declarative_base, sessionmaker, selectinload, relationship
from sqlalchemy import Integer, Column, ForeignKey, String
from sanic.response import json
from sqlalchemy.orm import sessionmaker, scoped_session
app = Sanic("my_app")
Base = declarative_base()
class BaseModel(Base):
__abstract__ = True
id = Column(Integer(), primary_key=True)
class Person(BaseModel):
__tablename__ = "person"
name = Column(String())
cars = relationship("Car")
def to_dict(self):
return {"name": self.name, "cars": [{"brand": car.brand} for car in self.cars]}
class Car(BaseModel):
__tablename__ = "car"
brand = Column(String())
user_id = Column(ForeignKey("person.id"))
user = relationship("Person", back_populates="cars")
# Create a new database file
database_file = "test.db"
bind = create_async_engine(f"sqlite+aiosqlite:///{database_file}", echo=True)
Session = sessionmaker(bind, class_=AsyncSession, expire_on_commit=False)
@app.middleware("request")
async def inject_session(request):
session = scoped_session(Session)()
request.ctx.session = session
request.ctx.session_ctx_token = session
from contextvars import ContextVar
session_ctx_var = ContextVar("session")
@app.middleware("request")
async def inject_session(request):
request.ctx.session = Session()
session_ctx_var.set(request.ctx.session)
@app.middleware("response")
async def close_session(request, response):
session = session_ctx_var.get()
if session is not None:
await session.close()
return response
@app.post("/user")
async def create_user(request):
session = request.ctx.session
async with session.begin():
car = Car(brand="Tesla")
person = Person(name="foo", cars=[car])
session.add_all([person])
return json(person.to_dict())
@app.get("/user/<pk:int>")
async def get_user(request, pk):
session = request.ctx.session
async with session.begin():
stmt = select(Person).where(Person.id == pk).options(selectinload(Person.cars))
result = await session.execute(stmt)
person = result.scalar()
if not person:
return json({})
return json(person.to_dict())
if __name__ == "__main__":
# Create tables
Base.metadata.create_all(bind)
app.run(host="0.0.0.0", port=8000)
4.1 Testing script
import requests
# Create a user
create_user_url = 'http://localhost:8000/user'
response = requests.post(create_user_url)
if response.status_code == 200:
created_user = response.json()
print("Created user:", created_user)
else:
print("Failed to create user")
# Retrieve a user
user_id = 1 # ID of the user to retrieve
get_user_url = f'http://localhost:8000/user/{user_id}'
response = requests.get(get_user_url)
if response.status_code == 200:
retrieved_user = response.json()
print("Retrieved user:", retrieved_user)
else:
print("Failed to retrieve user")
4.2 Results
C:\Users\Windows10>python
Python 3.10.5 (tags/v3.10.5:f377153, Jun 6 2022, 16:14:13) [MSC v.1929 64 bit (AMD64)] on win32
Type "help", "copyright", "credits" or "license" for more information.
>>> import requests
>>>
>>> # Create a user
>>> create_user_url = 'http://localhost:8000/user'
>>> response = requests.post(create_user_url)
>>> if response.status_code == 200:
... created_user = response.json()
... print("Created user:", created_user)
... else:
... print("Failed to create user")
...
Created user: {'name': 'foo', 'cars': [{'brand': 'Tesla'}]}
>>> # Retrieve a user
>>> user_id = 1 # ID of the user to retrieve
>>> get_user_url = f'http://localhost:8000/user/{user_id}'
>>> response = requests.get(get_user_url)
>>> if response.status_code == 200:
... retrieved_user = response.json()
... print("Retrieved user:", retrieved_user)
... else:
... print("Failed to retrieve user")
...
Retrieved user: {'name': 'foo', 'cars': [{'brand': 'Tesla'}]}
Assuming it’s idempotent, I’d move this script to a listener before startup. Then you don’t need to run anything. Whenever app starts you are assured to have the tables.
1 Like
Tortoise-ORM，there are examples on the official website 