Skip to content

AnswerDotAI/fastsql

BranchesTags

Repository files navigation

fastsql

Install

pip install fastsql

Overview

from fastcore.utils import *
from fastcore.net import urlsave
from fastsql import *
from fastsql.core import NotFoundError

We demonstrate fastsql‘s features here using the ’chinook’ sample database.

url = 'https://github.com/lerocha/chinook-database/raw/master/ChinookDatabase/DataSources/Chinook_Sqlite.sqlite'
path = Path('chinook.sqlite')
if not path.exists(): urlsave(url, path)
db = database("chinook.sqlite"); db
Database(sqlite:///chinook.sqlite)

Databases have a t property that lists all tables:

dt = db.t
dt
Album, Artist, Customer, Employee, Genre, Invoice, InvoiceLine, MediaType, Playlist, PlaylistTrack, Track

You can use this to grab a single table…:

# artist = dt.artists
# artist
artist = dt.Artist
artist
<Table Artist (ArtistId, Name)>

…or multiple tables at once:

dt['Artist','Album','Track','Genre','MediaType']
[<Table Artist (ArtistId, Name)>,
 <Table Album (AlbumId, Title, ArtistId)>,
 <Table Track (TrackId, Name, AlbumId, MediaTypeId, GenreId, Composer, Milliseconds, Bytes, UnitPrice)>,
 <Table Genre (GenreId, Name)>,
 <Table MediaType (MediaTypeId, Name)>]

It also provides auto-complete in Jupyter, IPython, and nearly any other interactive Python environment:

You can check if a table is in the database already:

'Artist' in dt
True

Column work in a similar way to tables, using the c property:

ac = artist.c
ac
ArtistId, Name

Auto-complete works for columns too:

Columns, tables, and view stringify in a format suitable for including in SQL statements. That means you can use auto-complete in f-strings.

qry = f"select * from {artist} where {ac.Name} like 'AC/%'"
print(qry)
select * from "Artist" where "Artist"."Name" like 'AC/%'

You can view the results of a select query using q:

db.q(qry)
[{'ArtistId': 1, 'Name': 'AC/DC'}]

Views can be accessed through the v property:

album = dt.Album

acca_sql = f"""select {album}.*
from {album} join {artist} using (ArtistId)
where {ac.Name} like 'AC/%'"""

db.create_view("AccaDaccaAlbums", acca_sql, replace=True)
acca_dacca = db.q(f"select * from {db.v.AccaDaccaAlbums}")
acca_dacca
[{'AlbumId': 1,
  'Title': 'For Those About To Rock We Salute You',
  'ArtistId': 1},
 {'AlbumId': 4, 'Title': 'Let There Be Rock', 'ArtistId': 1}]

Dataclass support

A dataclass type with the names, types, and defaults of the tables is created using dataclass():

album_dc = album.dataclass()
album_dc
fastsql.core.Album

Let’s try it:

album_obj = album_dc(**acca_dacca[0])
album_obj
Album(AlbumId=1, Title='For Those About To Rock We Salute You', ArtistId=1)

You can get the definition of the dataclass using fastcore’s dataclass_src – everything is treated as nullable, in order to handle auto-generated database values:

src = dataclass_src(album_dc)
hl_md(src, 'python')
@dataclass
class Album:
    AlbumId: int | None = UNSET
    Title: str | None = UNSET
    ArtistId: int | None = UNSET

Because dataclass() is dynamic, you won’t get auto-complete in editors like vscode – it’ll only work in dynamic environments like Jupyter and IPython. For editor support, you can export the full set of dataclasses to a module, which you can then import from:

create_mod(db, 'db_dc')
import sys
sys.path.insert(0, '.')
from db_dc import Track
Track()
Track(TrackId=UNSET, Name=UNSET, AlbumId=UNSET, MediaTypeId=UNSET, GenreId=UNSET, Composer=UNSET, Milliseconds=UNSET, Bytes=UNSET, UnitPrice=UNSET)

Indexing into a table does a query on primary key:

dt.Track[1]
Track(TrackId=1, Name='For Those About To Rock (We Salute You)', AlbumId=1, MediaTypeId=1, GenreId=1, Composer='Angus Young, Malcolm Young, Brian Johnson', Milliseconds=343719, Bytes=11170334, UnitPrice=Decimal('0.99'))

There’s a shortcut to select from a table – just call it as a function. If you’ve previously called dataclass(), returned iterms will be constructed using that class by default. There’s lots of params you can check out, such as limit:

album(limit=2)
[Album(AlbumId=1, Title='For Those About To Rock We Salute You', ArtistId=1),
 Album(AlbumId=2, Title='Balls to the Wall', ArtistId=2)]

Pass a truthy value as with_pk and you’ll get tuples of primary keys and records:

album(with_pk=1, limit=2)
[(1,
  Album(AlbumId=1, Title='For Those About To Rock We Salute You', ArtistId=1)),
 (2, Album(AlbumId=2, Title='Balls to the Wall', ArtistId=2))]

Indexing also uses the dataclass by default:

album[5]
Album(AlbumId=5, Title='Big Ones', ArtistId=3)

If you set xtra fields, then indexing is also filtered by those. As a result, for instance in this case, nothing is returned since album 5 is not created by artist 1:

album.xtra(ArtistId=1)

try: album[5]
except NotFoundError: print("Not found")
Not found

The same filtering is done when using the table as a callable:

album()
[Album(AlbumId=1, Title='For Those About To Rock We Salute You', ArtistId=1),
 Album(AlbumId=4, Title='Let There Be Rock', ArtistId=1)]

Core design

The following methods accept **kwargs, passing them along to the first dict param:

  • create
  • transform
  • transform_sql
  • update
  • insert
  • upsert
  • lookup

We can access a table that doesn’t actually exist yet:

dt
Album, Artist, Customer, Employee, Genre, Invoice, InvoiceLine, MediaType, Playlist, PlaylistTrack, Track

cats = dt.Cats
cats
<Table Cats (does not exist yet)>

We can use keyword arguments to now create that table:

cats.create(id=int, name=str, weight=float, uid=int, pk='id')
hl_md(cats.schema, 'sql')
CREATE TABLE "Cats" (
    id INTEGER, 
    name VARCHAR, 
    weight FLOAT, 
    uid INTEGER, 
    PRIMARY KEY (id)
)

It we set xtra then the additional fields are used for insert, update, and delete:

cats.xtra(uid=2)
cat = cats.insert(name='meow', weight=6)

The inserted row is returned, including the xtra ‘uid’ field.

cat
{'id': 1, 'name': 'meow', 'weight': 6.0, 'uid': 2}

Using ** in update here doesn’t actually achieve anything, since we can just pass a dict directly – it’s just to show that it works:

cat['name'] = "moo"
cat['uid'] = 1
cats.update(**cat)
cats()
[{'id': 1, 'name': 'moo', 'weight': 6.0, 'uid': 2}]

Attempts to update or insert with xtra fields are ignored.

An error is raised if there’s an attempt to update a record not matching xtra fields:

cats.xtra(uid=1)
try: cats.update(**cat)
except NotFoundError: print("Not found")
Not found

This all also works with dataclasses:

cats.xtra(uid=2)
cats.dataclass()
cat = cats[1]
cat
Cats(id=1, name='moo', weight=6.0, uid=2)

cats.drop()
cats
<Table Cats (id, name, weight, uid)>

Alternatively, you can create a table from a class. If it’s not already a dataclass, it will be converted into one. In either case, the dataclass will be created (or modified) so that None can be passed to any field (this is needed to support fields such as automatic row ids).

class Cat: id:int; name:str; weight:float; uid:int
cats = db.create(Cat)
hl_md(cats.schema, 'sql')
CREATE TABLE cat (
    id INTEGER, 
    name VARCHAR, 
    weight FLOAT, 
    uid INTEGER, 
    PRIMARY KEY (id)
)
cat = Cat(name='咪咪', weight=9)
cats.insert(cat)
Cat(id=1, name='咪咪', weight=9.0, uid=None)

Manipulating data

We try to make the following methods as flexible as possible. Wherever possible, they support Python dictionaries, dataclasses, and classes.

.insert()

Creates a record. Returns an instance of the updated record.

Insert using a dictionary.

cats.insert({'name': 'Rex', 'weight': 12.2})
Cat(id=2, name='Rex', weight=12.2, uid=None)

Insert using a dataclass.

CatDC = cats.dataclass()
cats.insert(CatDC(name='Tom', weight=10.2))
Cat(id=3, name='Tom', weight=10.2, uid=None)

Insert using a standard Python class

cat = cats.insert(Cat(name='Jerry', weight=5.2))

.update()

Updates a record using a Python dict, dataclass, or object, and returns an instance of the updated record.

Updating from a Python dict:

cats.update(dict(id=cat.id, name='Jerry', weight=6.2))
Cat(id=4, name='Jerry', weight=6.2, uid=None)

Updating from a dataclass:

cats.update(CatDC(id=cat.id, name='Jerry', weight=6.3))
Cat(id=4, name='Jerry', weight=6.3, uid=None)

Updating using a class:

cats.update(Cat(id=cat.id, name='Jerry', weight=5.7))
Cat(id=4, name='Jerry', weight=5.7, uid=None)

.delete()

Removing data is done by providing the primary key value of the record.

# Farewell Jerry!
cats.delete(cat.id)
Cat(id=4, name='Jerry', weight=5.7, uid=None)

Multi-field primary keys

Pass a collection of strings to create a multi-field pk:

class PetFood: catid:int; food:str; qty:int
petfoods = db.create(PetFood, pk=['catid','food'])
print(petfoods.schema)
CREATE TABLE pet_food (
    catid INTEGER, 
    food VARCHAR, 
    qty INTEGER, 
    PRIMARY KEY (catid, food)
)

You can index into these using multiple values:

pf = petfoods.insert(PetFood(1, 'tuna', 2))
petfoods[1,'tuna']
PetFood(catid=1, food='tuna', qty=2)

Updates work in the usual way:

pf.qty=3
petfoods.update(pf)
PetFood(catid=1, food='tuna', qty=3)

You can also use upsert to update if the key exists, or insert otherwise:

pf.qty=1
petfoods.upsert(pf)
petfoods()
[PetFood(catid=1, food='tuna', qty=1)]

pf.food='salmon'
petfoods.upsert(pf)
petfoods()
[PetFood(catid=1, food='tuna', qty=1), PetFood(catid=1, food='salmon', qty=1)]

delete takes a tuple of keys:

petfoods.delete((1, 'tuna'))
petfoods()
[PetFood(catid=1, food='salmon', qty=1)]

Migrations

FastSQL supports schema migrations to evolve your database over time. Migrations are SQL or Python files stored in a migrations directory, numbered sequentially.

The database tracks the current schema version in a _meta table. When you run migrations, only unapplied migrations are executed.

Let’s create a migration to add a priority field to our cats table:

# Create migrations directory
mig_dir = Path('cat_migrations')
mig_dir.mkdir(exist_ok=True)

# Create a migration to add priority column
migration_sql = 'alter table cat add column color text default "unknown";'
(mig_dir / '1-add_color_to_cat.sql').write_text(migration_sql)
56

Check the current schema version (will be 0 initially):

print(f"Current version: {db.version}")
Current version: 0

Run the migration:

db.migrate('cat_migrations')
Applied migration 1: 1-add_color_to_cat.sql

The database version is now updated, and the table structure reflects the change:

print(f"New version: {db.version}")
print(f"\nUpdated schema:")
cats = dt.cat
hl_md(cats.schema, 'sql')
New version: 1

Updated schema:

CREATE TABLE cat (
    id INTEGER, 
    name VARCHAR, 
    weight FLOAT, 
    uid INTEGER, 
    color TEXT DEFAULT "unknown", 
    PRIMARY KEY (id)
)

Existing records now have the priority field with the default value, and new records can use it too:

cats.insert({'name': 'Mr. Snuggles', 'weight': 8.5, 'color': 'tuxedo'})
cats()
[Cat(id=1, name='咪咪', weight=9.0, uid=None, color='unknown'),
 Cat(id=2, name='Rex', weight=12.2, uid=None, color='unknown'),
 Cat(id=3, name='Tom', weight=10.2, uid=None, color='unknown'),
 Cat(id=4, name='Mr. Snuggles', weight=8.5, uid=None, color='tuxedo')]

If you run migrate() again, it won’t reapply migrations that have already been applied:

db.migrate('cat_migrations')  # No output - migration already applied

Migrations can also be Python scripts. Create a file like 2-update_data.py that accepts the database connection string as a command line argument to perform more complex data transformations. Python migration scripts must handle their own commits:

# migrations/2-update_data.py
import sys
from fastsql import database

conn_str = sys.argv[1]
db = database(conn_str)

# Perform complex data transformations
for cat in db.t.cat():
    if cat.weight > 10:
        db.t.cat.update({'id': cat.id, 'priority': 1})

# Python migrations must commit their own changes
db.conn.commit()

About

answerdotai.github.io/fastsql

Resources

Readme

License

Apache-2.0 license
Activity
Custom properties

Stars

64 stars

Watchers

9 watching

Forks

5 forks
Report repository

Releases 5

v2.1.0 Latest
Jan 14, 2026
+ 4 releases

Contributors 6

Languages