1.22.0

Python 3.13 is now officially supported.

JSON / JSONB querying has been significantly improved. For example, if we have this table:

class RecordingStudio(Table):
    facilities = JSONB()

And the facilities column contains the following JSON data:

{
    "technicians": [
        {"name": "Alice Jones"},
        {"name": "Bob Williams"},
    ]
}

We can get the first technician name as follows:

>>> await RecordingStudio.select(
...     RecordingStudio.facilities["technicians"][0]["name"].as_alias("name")
... ).output(load_json=True)
[{'name': 'Alice Jones'}, ...]

TableStorage (used for dynamically creating Piccolo Table classes from an existing database) was improved, to support a Dockerised version of Piccolo Admin, which is coming soon.

1.21.0

Postgres 17 is now officially supported.

Fixed a bug with joins, when a ForeignKey column had db_column_name specified. Thanks to @jessemcl-flwls for reporting this issue.

1.20.0

get_related now works multiple layers deep:

concert = await Concert.objects().first()
manager = await concert.get_related(Concert.band_1._.manager)

1.19.1

Fixed a bug with the get_m2m method, which would raise a ValueError when no objects were found. It now handles this gracefully and returns an empty list instead. Thanks to @nVitius for this fix.

Improved the ASGI templates (including a fix for the latest Litestar version). Thanks to @sinisaos for this.

1.19.0

Added support for row locking (i.e. SELECT ... FOR UPDATE).

For example, if we have this table:

class Concert(Table):
    name = Varchar()
    tickets_available = Integer()

And we want to make sure that tickets_available never goes below 0, we can do the following:

async def book_tickets(ticket_count: int):
    async with Concert._meta.db.transaction():
        concert = await Concert.objects().where(
            Concert.name == "Awesome Concert"
        ).first().lock_rows()

        if concert.tickets_available >= ticket_count:
            await concert.update_self({
                Concert.tickets_available: Concert.tickets_available - ticket_count
            })
        else:
            raise ValueError("Not enough tickets are available!")

This means that when multiple transactions are running at the same time, it isn't possible to book more tickets than are available.

Thanks to @dkopitsa for adding this feature.

1.18.0

update_self

Added the update_self method, which is an alternative to the save method. Here's an example where it's useful:

# If we have a band object:
>>> band = await Band.objects().get(name="Pythonistas")
>>> band.popularity
1000

# We can increment the popularity, based on the current value in the
# database:
>>> await band.update_self({
...     Band.popularity: Band.popularity + 1
... })

# The new value is set on the object:
>>> band.popularity
1001

# It's safer than using the `save` method, because the popularity value on
# the object might be out of date with what's in the database:
band.popularity += 1
await band.save()

Thanks to @trondhindenes for suggesting this feature.

Batch raw queries

The batch method can now be used with raw queries. For example:

async with await MyTable.raw("SELECT * FROM my_table").batch() as batch:
    async for _batch in batch:
        print(_batch)

This is useful when you expect a raw query to return a lot of data.

Thanks to @devsarvesh92 for suggesting this feature.

1.17.1

Fixed a bug with migrations, where altering a column type from Integer to Float could fail. Thanks to @kurtportelli for reporting this issue.

1.17.0

Each migration is automatically wrapped in a transaction - this can now be disabled using the wrap_in_transaction argument:

manager = MigrationManager(
    wrap_in_transaction=False,
    ...
)

This is useful when writing a manual migration, and you want to manage all of the transaction logic yourself (or want multiple transactions).

granian is now a supported server in the ASGI templates. Thanks to @sinisaos for this.

1.16.0

Added custom async TestCase subclasses, to help with testing.

For example AsyncTransactionTest, which wraps each test in a transaction automatically:

class TestBandEndpoint(AsyncTransactionTest):

    async def test_band_response(self):
        """
        Make sure the endpoint returns a 200.
        """
        # This data automatically gets removed from the database when the
        # test finishes:
        band = Band({Band.name: "Pythonistas"})
        await band.save()

        # Using an API testing client, like httpx:
        response = await client.get(f"/bands/{band.id}/")
        self.assertEqual(response.status_code, 200)

And AsyncTableTest, which automatically creates and drops tables:

class TestBand(AsyncTableTest):

    # These tables automatically get created and dropped:
    tables = [Band]

    async def test_band(self):
        ...

1.15.0

Improved refresh - it now works with prefetched objects. For example:

>>> band = await Band.objects(Band.manager).first()
>>> band.manager.name
"Guido"

# If the manager has changed in the database, when we refresh the band, the
# manager object will also be updated:
>>> await band.refresh()
>>> band.manager.name
"New name"

Also, improved the error messages when creating a BaseUser - thanks to @haaavk for this.