Friction-free Cockroach DB support in Piccolo ORM.

PR: https://github.com/piccolo-orm/piccolo/pull/608

We'd like to see Cockroach Labs promote Piccolo as one of, if not the recommended async ORM. asyncpg has some support built-in to help us get there.

In the mean time I will be adding notes on how to use CockroachDB with Piccolo, and general progress.

Quick local setup for testing

• Download/extract (single binary): https://www.cockroachlabs.com/docs/stable/install-cockroachdb-linux.html
• Start Cockroach: ./cockroach start-single-node --insecure --store=node1 --listen-addr=localhost:26257 --http-addr=localhost:8080 --background
• Enter SQL client: ./cockroach sql --insecure
  • Can create database ...; use ...; from here.
• CTRL-C or killall cockroach to send graceful termination signal.

Default account is root without password. Fancy stats dashboard at http://localhost:8080

Both Piccolo Examples and Piccolo Admin are working.

What we still need

• [x] ~~Get build/test suite running on Github Actions for Piccolo QA tooling.~~
• [x] ~~Some way to identify we're running under Cockroach DB.~~
  • ~~Added CockroachDB Engine.~~
• [x] ~~Get test suite mostly / fully passing.~~
• [x] ~~Basic documentation~~
• [x] ~~Time Travel queries using AS OF SYSTEM TIME~~
  • ~~Some way to do Time Travel SELECT queries (SELECT ... AS OF SYSTEM TIME)~~
  • ~~https://www.cockroachlabs.com/docs/stable/as-of-system-time.html~~
  • :heavy_check_mark: See: https://github.com/piccolo-orm/piccolo/issues/607#issuecomment-1251330590
• [x] Sharding for Sequential Indexes
  • :warning: Done, but will go into its own PR after this one: https://github.com/piccolo-orm/piccolo/issues/622
  • Some way to do "Hash Sharded" Indexes (INDEX ... USING HASH)
  • https://www.cockroachlabs.com/docs/stable/hash-sharded-indexes.html

:bug: Known issues in CockroachDB or Asyncpg

[ ] ARRAY support between asyncpg ↔️ cockroach.

• Currently affected by: https://github.com/cockroachdb/cockroach/issues/71908
• See: piccolo/tests/columns/test_array.py
• M2M works, but as_list=True does not because it uses ARRAY

cannot drop UNIQUE constraint "..." using ALTER TABLE DROP CONSTRAINT, use DROP INDEX CASCADE instead

• Tracked: https://github.com/cockroachdb/cockroach/issues/42840
• Affecting test_alter_column_unique() in test_migrations.py

column ... is of type int[] and thus is not indexable / column ... is of type varchar[] and thus is not indexable

• Tracked: https://github.com/cockroachdb/cockroach/issues/35730
• Affecting both test_array_column_integer() and test_array_column_varchar in test_migration_manager.py

:moon: Future nice to haves

CockroachDB executes ALTER COLUMN queries asynchronously.

• In the short term this can be mitigated by keeping migrations small.
• At odds with Piccolo's assumption that the database is altered before the next migration operation begins. CockroachDB will give a warning: table <...> is currently undergoing a schema change if a later operation tries to modify the table before the asynchronous query finishes. CockroachDB may fix this in the future.

Fixes #107 #32 and maybe #63 Here's my approach to support UUID priamry key:

• Removed PrimaryKey column as it only supported Integer value types
• Didn't define id for Table by default and checked if no PrimaryKey was defined for table then Serial PK would be added.
• I think Serial column is Postgres specific, maybe change it to something like Django AutoField?

This works fine in Postgres, but in sqlite after insert we're returning cursor.lastrowid which does not return our id but the special virtual column ROWID from sqlite. I could come up with two approaches:

• Sqlite has added RETURNING clauses so we can do some sane INSERT query the same way as Postgres. But we will only support Sqlite 3.35 forwards: here
• Or we have to get the cursor.lastrowid and do another query to get our id based on ROWID like this SELECT id FROM table WHERE ROWID = <cursor.lastrowid>. I wanted to get some feedback on this.

Any of these should work now:

class MyTableDefaultPrimaryKey(Table):
    name = Varchar()


class MyTablePrimaryKeyInteger(Table):
    id = Integer(null=False, primary=True, key=True)
    name = Varchar()


class MyTablePrimaryKeyUUID(Table):
    id = UUID(null=False, primary=True, key=True)
    name = Varchar()

Hi, I often get sqlite3.OperationalError: database is locked, what would be the best way to evade this?

I thought about increasing the lock timeout, but am not sure how to correctly do this when using this library.

Here https://stackoverflow.com/a/39265148 the pragma-statement PRAGMA journal_mode=WAL is recommended. How would I use this with piccolo? I know I can execute raw sql, but doesn't this need to be re-executed for each connection?

Thank you for your help!

Piccolo now has a command for generating a schema from an existing database.

piccolo schema generate

The current implementation doesn't cover every possible edge case and database feature, so there's room for improvement. For example:

Column defaults

Getting the defaults for a column, and reflecting them in the Piccolo column. For example, if the column has a default of 1 in the database, we want that to be reflected in the Piccolo schema as:

class MyTable(Table):
    my_column = Integer(default=1)

On Delete / On Update

The database values aren't currently reflected in the ForeignKey column definitions.

class MyTable(Table):
    my_column = ForeignKey(on_delete=OnDelete.cascade)

Decimal precision

The precision of a decimal column in the database currently isn't reflected in the Decimal column definitions.

class MyTable(Table):
    my_column = Decimal(digits=(5,2))

It's a fairly beginner friendly feature to work on, because even though the code is fairly advanced, it's completely self contained, and doesn't require extensive knowledge of the rest of Piccolo.

resolves #344

here's my initial work on adding callbacks to queries as per #344

i would appreciate a code review before i take this any further! i also have a couple of questions:

1. at this stage you only get a callback when success = True, where & how would you suggest we detect a query failing and invoke the callback with success = False?
2. in piccolo/query/base.py, at the end of the run function, it branches based on whether there's more than one querystring or not. i was unable to cover the second branch with a test as i couldn't figure out how you'd end up with more than one querystring. would appreciate some pointers on how that bit works.

i look forward to hearing your thoughts.

• [x] JSON fields
• [ ] More aggregation functions - e.g. SUM
• [x] Aliases in SELECT queries (e.g. SELECT foo AS bar FROM my_table)
• [ ] Improved support for data migrations
• [x] Fixtures
• [x] Schema visualisation tool
• [x] piccolo sql_shell new command - execute SQL directly in a shell, via the configured engine
• [x] piccolo shell new command - like the playground, but for your own tables
• [x] Shorter aliases for commands in CLI, or tab completion
• [x] Improved documentation for how to test Piccolo apps
• [ ] Nested transactions
• [x] Move Pydantic integration from Piccolo API into Piccolo
• [x] Allow the ForeignKey references argument to accept a string such as 'my_app.SomeTable' when circular imports are an issue.
• [ ] Support Postgres Enum types.
• [ ] Row level constraints.
• [ ] Making foreign key constraints optional on ForeignKey columns.
• [x] Allow UUIDs as primary keys?
• [ ] Subqueries

I think it'd be clear to move all the scripts to the scripts folder following this. I've moved all existing scripts and updated the github workflow. But here some questions/ideas:

• Should we move piccolo.sh from root of project too?
• Coverage is not enforced, should we add a minimum coverage so pipeline fails on missing coverage using something like Codecov?
• release.sh can be a workflow in github to upload to PyPi when github release is created.

Hello again, I tried out PostgresEngine but noticed that using camelCase column names leads to problems. Is this something the ORM could potentially "abstract away", or would that be a bad idea?

An initial prototype for M2M (many to many). It still needs a bunch of testing, but it's a start.

class Band(Table):
    name = Varchar()
    genres = M2M(
        LazyTableReference("GenreToBand", module_path=__name__)
    )


class Genre(Table):
    name = Varchar()


# The joining table:
class GenreToBand(Table):
    band = ForeignKey(Band)
    genre = ForeignKey(Genre)

We have a new class called M2M. You reference a joining table. By explicitly defining the joining table it has the following benefits:

• Simplifies migrations, as we don't have to magically create new tables.
• Makes it easier to retrofit Piccolo onto an existing schema.
• Can add additional columns to the joining table if we want (see below).

class GenreToBand(Table):
    band = ForeignKey(Band)
    genre = ForeignKey(Genre)
    created_on = Timestamp()
    reasoning = Text()

With M2M you can then do queries like this:

>>> await Band.select(Band.name, Band.genres(Genre.name))
[
    {"name": "Pythonistas", "genres": ["Rock", "Folk"]},
    {"name": "Rustaceans", "genres": ["Folk"]},
    {"name": "C-Sharps", "genres": ["Rock", "Classical"]},
]

From piccolo shell run the commands

Book.select().limit(4).run_sync()

and

Book.select().offset(1).limit(4).run_sync()

work fine, but...

Book.select().limit(5).run_sync()

displays the result and then raise BlockingIOError exception:

~/.venv/lib/python3.9/site-packages/IPython/core/displayhook.py in write_format_data(self, format_dict, md_dict)
    189
    190         try:
--> 191             print(result_repr)
    192         except UnicodeEncodeError:
    193             # If a character is not supported by the terminal encoding replace

BlockingIOError: [Errno 35] write could not complete without blocking---------------------------------------------------------------------------
BlockingIOError                           Traceback (most recent call last)

I don't know what's going on.

get_output_schema will raise an exception with the name of the corresponding table if something goes wrong while generating the tables. based on #447

Another alternative would be to add the errors attribute to the OutputSchema class and handle it like how warnings are being addressed.

In django each unit test is wrapped in a database transaction. It means that any changes made during test (e.g. calling an endpoint which adds a new object) are rolled back after the test ends.

This is very useful because the state of the DB does not change between tests, and assertions are easy to write and very easy to understand (like: I call an endpoint to create an object, then I check that there is exactly 1 object in the database).

Is there a way of achieving the same transactionality in piccolo and FastAPI ? Are there any examples?

As far as I can see, if using Readable and get_readable, boolean fields are represented as f and t (False and True). It would be good if it was possible to represent these in a more user-friendly way - Maybe even make the default similar to Python's string representation of booleans:

In [1]: str(True)
Out[1]: 'True'

In [2]: str(False)
Out[2]: 'False'

Why

below I created a mock application with typer to illustrate my point of view:

• better information display

• better error display :

• type checking everywhere
• modularity
• shell completion
• all the powerful stuff that rich provides ( i.e tables, spinners, progress bars,...)

using questionary provides a much more intuitive user experience:

If any of the maintainers finds this interesting, you can assign it to me and I will work on it 🙂

I have following tables.

class Status(Table):
    code = SmallInt()
    name = Varchar()

class Report(Table):
    id = UUID(primary_key=True)
    form = Varchar()
    # columns

class Message(Table):
    id = UUID(primary_key=True)
    report_id = ForeignKey(references=Report)
    status = ForeignKey(references=Status, target_column=Status.code)
    # columns

class Client(Table):
    report_id = ForeignKey(references=Report)
    type = SmallInt()
    # columns

I want to make this query.

SELECT message.id, message.report_id, report.form, status.name, client.type FROM message LEFT JOIN report ON report.id = message.report_id LEFT JOIN status ON status.code = message.status LEFT JOIN client ON client.report_id = message.report_id WHERE message.status != 0

This variant returns a rather obvious error.

result = await Message.select(
    Message.id,
    Message.report_id,
    Message.report_id.form,
    Message.status.name,
    Message.report_id.client.type  # ?? AttributeError: 'ForeignKey' object has no attribute 'client'
).where(
    Message.status.code != 0
)

How can I join a table with client details?

Hello! I've using piccolo with fastapi and pydantic. I need to select() db records as dicts and throw them to pydantic, but it fails on ForeignKey because piccolo returns nested dicts with nested=True param as dicts with filled keys and None values

{'id': 1, 'one': {'desc': None, 'id': None, 'name': None, 'number': None}}

instead of just

{'id': 1, 'one': None}

import asyncio
from typing import Optional

import piccolo.columns as cl
from piccolo.engine.sqlite import SQLiteEngine
from piccolo.table import Table
from pydantic import BaseModel


DB = SQLiteEngine()


class TableOne(Table, db=DB):
    name = cl.Varchar()
    desc = cl.Varchar()
    number = cl.Integer()


class TableTwo(Table, db=DB):
    one = cl.ForeignKey(TableOne)


class TableOneModel(BaseModel):
    id: int
    name: str
    desc: str
    number: int


class TableTwoModel(BaseModel):
    id: int
    one: Optional[TableOneModel] = None


async def main():
    await TableOne.alter().drop_table(if_exists=True)
    await TableOne.create_table()
    await TableTwo.alter().drop_table(if_exists=True)
    await TableTwo.create_table()

    await TableTwo.objects().create()

    result = await TableTwo.select(
        TableTwo.all_columns(), TableTwo.one.all_columns()
    ).output(nested=True).first()
    print(result)
    # fails here
    TableTwoModel(**result)


if __name__ == '__main__':
    asyncio.run(main())