API reference¶
Table¶
- class piccolo.table.Table(_data: Dict[Column, Any] | None = None, _ignore_missing: bool = False, _exists_in_db: bool = False, **kwargs)¶
The class represents a database table. An instance represents a row.
The constructor can be used to assign column values.
Note
The
_data
,_ignore_missing
, and_exists_in_db
arguments are prefixed with an underscore to help prevent a clash with a column name which might be passed in via kwargs.- Parameters:
_data –
There’s two ways of passing in the data for each column. Firstly, you can use kwargs:
Band(name="Pythonistas")
Secondly, you can pass in a dictionary which maps column classes to values:
Band({Band.name: 'Pythonistas'})
The advantage of this second approach is it’s more strongly typed, and linters such as flake8 or MyPy will more easily detect typos.
_ignore_missing – If
False
aValueError
will be raised if any column values haven’t been provided._exists_in_db – Used internally to track whether this row exists in the database.
- add_m2m(*rows: Table, m2m: M2M, extra_column_values: Dict[Column | str, Any] = {}) M2MAddRelated ¶
Save the row if it doesn’t already exist in the database, and insert an entry into the joining table.
>>> band = await Band.objects().get(Band.name == "Pythonistas") >>> await band.add_m2m( ... Genre(name="Punk rock"), ... m2m=Band.genres ... ) [{'id': 1}]
- Parameters:
extra_column_values –
If the joining table has additional columns besides the two required foreign keys, you can specify the values for those additional columns. For example, if this is our joining table:
class GenreToBand(Table): band = ForeignKey(Band) genre = ForeignKey(Genre) reason = Text()
We can provide the
reason
value:await band.add_m2m( Genre(name="Punk rock"), m2m=Band.genres, extra_column_values={ "reason": "Their second album was very punk." } )
- classmethod all_columns(exclude: Sequence[str | Column] | None = None) List[Column] ¶
Used in conjunction with
select
queries. Just as we can useall_columns
to retrieve all of the columns from a related table, we can also use it at the root of our query to get all of the columns for the root table. For example:await Band.select( Band.all_columns(), Band.manager.all_columns() )
This is mostly useful when the table has a lot of columns, and typing them out by hand would be tedious.
- Parameters:
exclude – You can request all columns, except these.
Used in conjunction with
objects
queries. Just as we can useall_related
on aForeignKey
, you can also use it for the table at the root of the query, which will return each related row as a nested object. For example:concert = await Concert.objects( Concert.all_related() ) >>> concert.band_1 <Band: 1> >>> concert.band_2 <Band: 2> >>> concert.venue <Venue: 1>
This is mostly useful when the table has a lot of foreign keys, and typing them out by hand would be tedious. It’s equivalent to:
concert = await Concert.objects( Concert.venue, Concert.band_1, Concert.band_2 )
- Parameters:
exclude – You can request all columns, except these.
- classmethod alter() Alter ¶
Used to modify existing tables and columns.
await Band.alter().rename_column(Band.popularity, 'rating')
- classmethod count(column: Column | None = None, distinct: Sequence[Column] | None = None) Count ¶
Count the number of matching rows:
await Band.count().where(Band.popularity > 1000)
- Parameters:
column – If specified, just count rows where this column isn’t null.
distinct –
Counts the number of distinct values for these columns. For example, if we have a concerts table:
class Concert(Table): band = Varchar() start_date = Date()
With this data:
band
start_date
Pythonistas
2023-01-01
Pythonistas
2023-02-03
Rustaceans
2023-01-01
Without the
distinct
argument, we get the count of all rows:>>> await Concert.count() 3
To get the number of unique concert dates:
>>> await Concert.count(distinct=[Concert.start_date]) 2
- classmethod create_index(columns: List[Column] | List[str], method: IndexMethod = IndexMethod.btree, if_not_exists: bool = False) CreateIndex ¶
Create a table index. If multiple columns are specified, this refers to a multicolumn index, rather than multiple single column indexes.
await Band.create_index([Band.name])
- classmethod create_table(if_not_exists=False, only_default_columns=False, auto_create_schema: bool = True) Create ¶
Create table, along with all columns.
await Band.create_table()
- classmethod delete(force=False) Delete ¶
Delete rows from the table.
await Band.delete().where(Band.name == 'Pythonistas')
- Parameters:
force – Unless set to
True
, deletions aren’t allowed without awhere
clause, to prevent accidental mass deletions.
- classmethod drop_index(columns: List[Column] | List[str], if_exists: bool = True) DropIndex ¶
Drop a table index. If multiple columns are specified, this refers to a multicolumn index, rather than multiple single column indexes.
await Band.drop_index([Band.name])
- classmethod exists() Exists ¶
Use it to check if a row exists, not if the table exists.
await Band.exists().where(Band.name == 'Pythonistas')
- classmethod from_dict(data: Dict[str, Any]) TableInstance ¶
Used when loading fixtures. It can be overriden by subclasses in case they have specific logic / validation which needs running when loading fixtures.
- get_m2m(m2m: M2M) M2MGetRelated ¶
Get all matching rows via the join table.
>>> band = await Band.objects().get(Band.name == "Pythonistas") >>> await band.get_m2m(Band.genres) [<Genre: 1>, <Genre: 2>]
- classmethod get_readable() Readable ¶
Creates a readable representation of the row.
- get_related(foreign_key: str) GetRelated[Table]
Used to fetch a
Table
instance, for the target of a foreign key.band = await Band.objects().first() manager = await band.get_related(Band.manager) >>> print(manager.name) 'Guido'
It can only follow foreign keys multiple levels deep. For example,
Concert.band_1.manager
.
- classmethod indexes() Indexes ¶
Returns a list of the indexes for this tables.
await Band.indexes()
- classmethod insert(*rows: TableInstance) Insert[TableInstance] ¶
Insert rows into the database.
await Band.insert( Band(name="Pythonistas", popularity=500, manager=1) )
- classmethod objects(*prefetch: ForeignKey | List[ForeignKey]) Objects[TableInstance] ¶
Returns a list of table instances (each representing a row), which you can modify and then call ‘save’ on, or can delete by calling ‘remove’.
pythonistas = await Band.objects().where( Band.name == 'Pythonistas' ).first() pythonistas.name = 'Pythonistas Reborn' await pythonistas.save() # Or to remove it from the database: await pythonistas.remove()
- Parameters:
prefetch –
Rather than returning the primary key value of this related table, a nested object will be returned for the row on the related table.
# Without nested band = await Band.objects().first() >>> band.manager 1 # With nested band = await Band.objects(Band.manager).first() >>> band.manager <Band 1>
- property querystring: QueryString¶
Used when inserting rows.
- classmethod raw(sql: str, *args: Any) Raw ¶
Execute raw SQL queries on the underlying engine - use with caution!
await Band.raw('select * from band')
Or passing in parameters:
await Band.raw("SELECT * FROM band WHERE name = {}", 'Pythonistas')
- classmethod ref(column_name: str) Column ¶
Used to get a copy of a column from a table referenced by a
ForeignKey
column. It’s unlikely an end user of this library will ever need to do this, but other libraries built on top of Piccolo may need this functionality.Band.ref('manager.name')
- refresh(columns: Sequence[Column] | None = None, load_json: bool = False) Refresh ¶
Used to fetch the latest data for this instance from the database. Modifies the instance in place, but also returns it as a convenience.
- Parameters:
columns – If you only want to refresh certain columns, specify them here. Otherwise all columns are refreshed.
load_json – Whether to load
JSON
/JSONB
columns as objects, instead of just a string.
Example usage:
# Get an instance from the database. instance = await Band.objects.first() # Later on we can refresh this instance with the latest data # from the database, in case it has gotten stale. await instance.refresh() # Alternatively, running it synchronously: instance.refresh().run_sync()
- remove() Delete ¶
A proxy to a delete query.
- remove_m2m(*rows: Table, m2m: M2M) M2MRemoveRelated ¶
Remove the rows from the joining table.
>>> band = await Band.objects().get(Band.name == "Pythonistas") >>> genre = await Genre.objects().get(Genre.name == "Rock") >>> await band.remove_m2m( ... genre, ... m2m=Band.genres ... )
- save(columns: Sequence[str | Column] | None = None) Insert | Update ¶
A proxy to an insert or update query.
- Parameters:
columns –
Only the specified columns will be synced back to the database when doing an update. For example:
band = await Band.objects().first() band.popularity = 2000 await band.save(columns=[Band.popularity])
If
columns=None
(the default) then all columns will be synced back to the database.
- classmethod select(*columns: t.Union[Selectable, str], exclude_secrets=False) Select ¶
Get data in the form of a list of dictionaries, with each dictionary representing a row.
These are all equivalent:
await Band.select().columns(Band.name) await Band.select(Band.name) await Band.select('name')
- Parameters:
exclude_secrets – If
True
, any columns withsecret=True
are omitted from the response. For example, we use this for the password column ofBaseUser
. Even though the passwords are hashed, you still don’t want them being passed over the network if avoidable.
- classmethod table_exists() TableExists ¶
Check if the table exists in the database.
await Band.table_exists()
- to_dict(*columns: Column) Dict[str, Any] ¶
A convenience method which returns a dictionary, mapping column names to values for this table instance.
instance = await Manager.objects().get( Manager.name == 'Guido' ) >>> instance.to_dict() {'id': 1, 'name': 'Guido'}
If the columns argument is provided, only those columns are included in the output. It also works with column aliases.
>>> instance.to_dict(Manager.id, Manager.name.as_alias('title')) {'id': 1, 'title': 'Guido'}
- classmethod update(values: Dict[Column | str, Any] | None = None, force: bool = False, use_auto_update: bool = True, **kwargs) Update ¶
Update rows.
All of the following work, though the first is preferable:
await Band.update( {Band.name: "Spamalot"} ).where( Band.name == "Pythonistas" ) await Band.update( {"name": "Spamalot"} ).where( Band.name == "Pythonistas" ) await Band.update( name="Spamalot" ).where( Band.name == "Pythonistas" )
- Parameters:
force – Unless set to
True
, updates aren’t allowed without awhere
clause, to prevent accidental mass overriding of data.use_auto_update – Whether to use the
auto_update
values on any columns. See theauto_update
argument onColumn
for more information.
- update_self(values: Dict[Column | str, Any]) UpdateSelf ¶
This allows the user to update a single object - useful when the values are derived from the database in some way.
For example, if we have the following table:
class Band(Table): name = Varchar() popularity = Integer()
And we fetch an object:
>>> band = await Band.objects().get(name="Pythonistas")
We could use the typical syntax for updating the object:
>>> band.popularity += 1 >>> await band.save()
The problem with this, is what if another object has already incremented
popularity
? It would overide the value.Instead we can do this:
>>> await band.update_self({ ... Band.popularity: Band.popularity + 1 ... })
This updates
popularity
in the database, and also sets the new value forpopularity
on the object.
SchemaManager¶
- class piccolo.schema.SchemaManager(db: Engine | None = None)¶
A useful utility class for interacting with schemas.
- Parameters:
db – Used to execute the database queries. If not specified, we try and import it from
piccolo_conf.py
.
- create_schema(schema_name: str, *, if_not_exists: bool = True) CreateSchema ¶
Creates the specified schema:
>>> await SchemaManager().create_schema(schema_name="music")
- Parameters:
schema_name – The name of the schema to create.
if_not_exists – No error will be raised if the schema already exists.
- drop_schema(schema_name: str, *, if_exists: bool = True, cascade: bool = False) DropSchema ¶
Drops the specified schema:
>>> await SchemaManager().drop_schema(schema_name="music")
- Parameters:
schema_name – The name of the schema to drop.
if_exists – No error will be raised if the schema doesn’t exist.
cascade – If
True
then it will automatically drop the tables within the schema.
- list_schemas() ListSchemas ¶
Returns the name of each schema in the database:
>>> await SchemaManager().list_schemas() ['public', 'schema_1']
- list_tables(schema_name: str) ListTables ¶
Returns the name of each table in the given schema:
>>> await SchemaManager().list_tables(schema_name="music") ['band', 'manager']
- Parameters:
schema_name – List the tables in this schema.
- move_table(table_name: str, new_schema: str, current_schema: str | None = None) MoveTable ¶
Moves a table to a different schema:
>>> await SchemaManager().move_schema( ... table_name='my_table', ... new_schema='schema_1' ... )
- Parameters:
table_name – The name of the table to move.
new_schema – The name of the scheam you want to move the table too.
- Current_schema:
If not specified,
'public'
is assumed.
Column¶
- class piccolo.columns.base.Column(null: bool = False, primary_key: bool = False, unique: bool = False, index: bool = False, index_method: IndexMethod = IndexMethod.btree, required: bool = False, help_text: str | None = None, choices: Type[Enum] | None = None, db_column_name: str | None = None, secret: bool = False, auto_update: Any = Ellipsis, **kwargs)¶
All other columns inherit from
Column
. Don’t use it directly.The following arguments apply to all column types:
- Parameters:
null – Whether the column is nullable.
primary_key – If set, the column is used as a primary key.
default – The column value to use if not specified by the user.
unique – If set, a unique constraint will be added to the column.
index – Whether an index is created for the column, which can improve the speed of selects, but can slow down inserts.
index_method – If index is set to
True
, this specifies what type of index is created.required – This isn’t used by the database - it’s to indicate to other tools that the user must provide this value. Example uses are in serialisers for API endpoints, and form fields.
help_text – This provides some context about what the column is being used for. For example, for a
Decimal
column calledvalue
, it could say'The units are millions of dollars'
. The database doesn’t use this value, but tools such as Piccolo Admin use it to show a tooltip in the GUI.choices – An optional Enum - when specified, other tools such as Piccolo Admin will render the available options in the GUI.
db_column_name –
If specified, you can override the name used for the column in the database. The main reason for this is when using a legacy database, with a problematic column name (for example
'class'
, which is a reserved Python keyword). Here’s an example:class MyTable(Table): class_ = Varchar(db_column_name="class") >>> await MyTable.select(MyTable.class_) [{'id': 1, 'class': 'test'}]
This is an advanced feature which you should only need in niche situations.
secret –
If
secret=True
is specified, it allows a user to automatically omit any fields when doing a select query, to help prevent inadvertent leakage of sensitive data.class Band(Table): name = Varchar() net_worth = Integer(secret=True) >>> await Band.select(exclude_secrets=True) [{'name': 'Pythonistas'}]
auto_update –
Allows you to specify a value to set this column to each time it is updated (via
MyTable.update
, orMyTable.save
on an existing row). A common use case is having amodified_on
column.class Band(Table): name = Varchar() popularity = Integer() # The value can be a function or static value: modified_on = Timestamp(auto_update=datetime.datetime.now) # This will automatically set the `modified_on` column to the # current timestamp, without having to explicitly set it: >>> await Band.update({ ... Band.popularity: Band.popularity + 100 ... }).where(Band.name == 'Pythonistas')
Note - this feature is implemented purely within the ORM. If you want similar functionality on the database level (i.e. if you plan on using raw SQL to perform updates), then you may be better off creating SQL triggers instead.
- as_alias(name: str) Column ¶
Allows column names to be changed in the result of a select.
For example:
>>> await Band.select(Band.name.as_alias('title')).run() {'title': 'Pythonistas'}
- get_default_value() Any ¶
If the column has a default attribute, return it. If it’s callable, return the response instead.
- get_select_string(engine_type: str, with_alias: bool = True) QueryString ¶
How to refer to this column in a SQL query, taking account of any joins and aliases.
- get_sql_value(value: Any, delimiter: str = "'") str ¶
When using DDL statements, we can’t parameterise the values. An example is when setting the default for a column. So we have to convert from the Python type to a string representation which we can include in our DDL statements.
- Parameters:
value – The Python value to convert to a string usable in a DDL statement e.g.
1
.delimiter – The string returned by this function is wrapped in delimiters, ready to be added to a DDL statement. For example:
'hello world'
.
- Returns:
The string usable in the DDL statement e.g.
'1'
.
- ilike(value: str) Where ¶
Only Postgres supports ILIKE. It’s used for case insensitive matching.
For SQLite, it’s just proxied to a LIKE query instead.
- is_not_null() Where ¶
Can be used instead of
MyTable.column != None
, because some linters don’t like a comparison toNone
.
- is_null() Where ¶
Can be used instead of
MyTable.column == None
, because some linters don’t like a comparison toNone
.
- join_on(column: Column) ForeignKey ¶
Joins are typically performed via foreign key columns. For example, here we get the band’s name and the manager’s name:
class Manager(Table): name = Varchar() class Band(Table): name = Varchar() manager = ForeignKey(Manager) >>> await Band.select(Band.name, Band.manager.name)
The
join_on
method lets you join tables even when foreign keys don’t exist, by joining on a column in another table.For example, here we want to get the manager’s email, but no foreign key exists:
class Manager(Table): name = Varchar(unique=True) email = Varchar() class Band(Table): name = Varchar() manager_name = Varchar() >>> await Band.select( ... Band.name, ... Band.manager_name.join_on(Manager.name).email ... )
- like(value: str) Where ¶
Both SQLite and Postgres support LIKE, but they mean different things.
In Postgres, LIKE is case sensitive (i.e. ‘foo’ equals ‘foo’, but ‘foo’ doesn’t equal ‘Foo’).
In SQLite, LIKE is case insensitive for ASCII characters (i.e. ‘foo’ equals ‘Foo’). But not for non-ASCII characters. To learn more, see the docs:
https://sqlite.org/lang_expr.html#the_like_glob_regexp_and_match_operators
Aggregate functions¶
Count¶
- class piccolo.query.methods.select.Count(column: Column | None = None, distinct: Sequence[Column] | None = None, alias: str = 'count')¶
Used in
Select
queries, usually in conjunction with thegroup_by
clause:>>> await Band.select( ... Band.manager.name.as_alias('manager_name'), ... Count(alias='band_count') ... ).group_by(Band.manager) [{'manager_name': 'Guido', 'count': 1}, ...]
It can also be used without the
group_by
clause (though you may prefer to theTable.count
method instead, as it’s more convenient):>>> await Band.select(Count()) [{'count': 3}]
- Parameters:
column – If specified, the count is for non-null values in that column.
distinct – If specified, the count is for distinct values in those columns.
alias –
The name of the value in the response:
# These two are equivalent: await Band.select( Band.name, Count(alias="total") ).group_by(Band.name) await Band.select( Band.name, Count().as_alias("total") ).group_by(Band.name)
Refresh¶
- class piccolo.query.methods.refresh.Refresh(instance: Table, columns: t.Optional[t.Sequence[Column]] = None, load_json: bool = False)¶
Used to refresh
Table
instances with the latest data data from the database. Accessible viarefresh
.- Parameters:
instance – The instance to refresh.
columns – Which columns to refresh - it not specified, then all columns are refreshed.
load_json – Whether to load
JSON
/JSONB
columns as objects, instead of just a string.
LazyTableReference¶
- class piccolo.columns.LazyTableReference(table_class_name: str, app_name: str | None = None, module_path: str | None = None)¶
Holds a reference to a
Table
subclass. Used to avoid circular dependencies in thereferences
argument ofForeignKey
columns.- Parameters:
table_class_name – The name of the
Table
subclass. For example,'Manager'
.app_name – If specified, the
Table
subclass is imported from a Piccolo app with the given name.module_path –
If specified, the
Table
subclass is imported from this path. For example,'my_app.tables'
.Hint
If the table is in the same file, you can pass in
__name__
.
Enums¶
Foreign Keys¶
- class piccolo.columns.OnDelete(value, names=None, *, module=None, qualname=None, type=None, start=1, boundary=None)¶
Used by
ForeignKey
to specify the behaviour when a related row is deleted.- cascade = 'CASCADE'¶
- no_action = 'NO ACTION'¶
- restrict = 'RESTRICT'¶
- set_default = 'SET DEFAULT'¶
- set_null = 'SET NULL'¶
- class piccolo.columns.OnUpdate(value, names=None, *, module=None, qualname=None, type=None, start=1, boundary=None)¶
Used by
ForeignKey
to specify the behaviour when a related row is updated.- cascade = 'CASCADE'¶
- no_action = 'NO ACTION'¶
- restrict = 'RESTRICT'¶
- set_default = 'SET DEFAULT'¶
- set_null = 'SET NULL'¶
Indexes¶
Column defaults¶
Date¶
- class piccolo.columns.defaults.DateOffset(days: int)¶
This makes the default value for a
Date
column the current date, but offset by a number of days.For example, if you wanted the default to be tomorrow, you can specify
DateOffset(days=1)
:class DiscountCode(Table): expires = Date(default=DateOffset(days=1))
- Parameters:
days – The number of days to offset.
Testing¶
ModelBuilder¶
- class piccolo.testing.model_builder.ModelBuilder¶
- async classmethod build(table_class: Type[TableInstance], defaults: Dict[Column | str, Any] | None = None, persist: bool = True, minimal: bool = False) TableInstance ¶
Build a
Table
instance with random data and save async. If theTable
has any foreign keys, then the related rows are also created automatically.- Parameters:
table_class – Table class to randomize.
defaults – Any values specified here will be used instead of random values.
persist – Whether to save the new instance in the database.
minimal – If
True
then any columns withnull=True
are assigned a value ofNone
.
Examples:
# Create a new instance with all random values: manager = await ModelBuilder.build(Manager) # Create a new instance, with certain defaults: manager = await ModelBuilder.build( Manager, {Manager.name: 'Guido'} ) # Create a new instance, but don't save it in the database: manager = await ModelBuilder.build(Manager, persist=False) # Create a new instance, with all null values set to None: manager = await ModelBuilder.build(Manager, minimal=True) # We can pass other table instances in as default values: band = await ModelBuilder.build(Band, {Band.manager: manager})
create_db_tables / drop_db_tables¶
- async piccolo.table.create_db_tables(*tables: Type[Table], if_not_exists: bool = False) None ¶
Creates the database table for each
Table
class passed in. The tables are created in the correct order, based on their foreign keys.- Parameters:
tables – The tables to create in the database.
if_not_exists – No errors will be raised if any of the tables already exist in the database.
- piccolo.table.create_db_tables_sync(*tables: Type[Table], if_not_exists: bool = False) None ¶
A sync wrapper around
create_db_tables()
.
- async piccolo.table.drop_db_tables(*tables: Type[Table]) None ¶
Drops the database table for each
Table
class passed in. The tables are dropped in the correct order, based on their foreign keys.- Parameters:
tables – The tables to delete from the database.
- piccolo.table.drop_db_tables_sync(*tables: Type[Table]) None ¶
A sync wrapper around
drop_db_tables()
.