flowgrate/python

Python SDK for Flowgrate — Laravel-style database migrations with a fluent API.

How it works

Define migrations in Python using the fluent Blueprint API. The SDK serializes them to JSON and pipes to the Flowgrate CLI, which compiles and executes the SQL.

Requirements

• Python 3.10+
• Flowgrate CLI — download the binary for your platform and put it on your PATH

# Linux (amd64)
curl -L https://github.com/flowgrate/core/releases/latest/download/flowgrate-linux-amd64 -o flowgrate
chmod +x flowgrate
sudo mv flowgrate /usr/local/bin/

# macOS (Apple Silicon)
curl -L https://github.com/flowgrate/core/releases/latest/download/flowgrate-darwin-arm64 -o flowgrate
chmod +x flowgrate
sudo mv flowgrate /usr/local/bin/

# macOS (Intel)
curl -L https://github.com/flowgrate/core/releases/latest/download/flowgrate-darwin-amd64 -o flowgrate
chmod +x flowgrate
sudo mv flowgrate /usr/local/bin/

# Or build from source
go install github.com/flowgrate/core@latest

Setup

1. Install the SDK:

pip install flowgrate

2. Create Program.py (entry point for the CLI to invoke):

from flowgrate import FlowgrateRunner
import os

FlowgrateRunner.run(os.path.dirname(__file__))

3. Create flowgrate.yml next to your migrations:

Generate it with the CLI (recommended):

flowgrate init --db=postgres://user:pass@localhost/mydb

Or create manually:

database:
  url: postgres://user:pass@localhost/mydb

migrations:
  project: ./migrations
  sdk: python
  run: python Program.py

4. Generate and run migrations:

flowgrate make CreateUsersTable
flowgrate up

Migration anatomy

from flowgrate import Migration, Schema


class CreateUsersTable(Migration):
    def up(self):
        with Schema.create("users") as t:
            t.id()
            t.string("name")
            t.string("email", 100)
            t.timestamps()

    def down(self):
        Schema.drop_if_exists("users")

Migration files must follow the naming convention: YYYYMMDD_HHMMSS_MigrationName.py

Blueprint API reference

Create / drop table

Schema.create("users")          # CREATE TABLE
Schema.table("users")           # ALTER TABLE
Schema.drop("users")            # DROP TABLE
Schema.drop_if_exists("users")  # DROP TABLE IF EXISTS

Column types

t.id()                          # BIGSERIAL PRIMARY KEY
t.small_integer("level")        # SMALLINT
t.integer("views")              # INTEGER
t.big_integer("score")          # BIGINT
t.decimal("price", 10, 2)       # NUMERIC(10, 2)
t.float("rating")               # REAL
t.double("latitude")            # DOUBLE PRECISION
t.boolean("active")             # BOOLEAN
t.string("name")                # VARCHAR(255)
t.string("code", 10)            # VARCHAR(10)
t.text("bio")                   # TEXT
t.uuid("public_id")             # UUID
t.json("settings")              # JSON
t.jsonb("metadata")             # JSONB
t.binary("avatar")              # BYTEA
t.date("birthday")              # DATE
t.time("opens_at")              # TIME
t.timestamp("verified_at")      # TIMESTAMP

Column modifiers (chainable)

.nullable()                     # NULL
.default(value)                 # DEFAULT value
.default_expression("NOW()")    # DEFAULT NOW()  — raw SQL
.generated_uuid()               # DEFAULT gen_random_uuid()
.comment("description")
.unique()                       # single-column unique index

Helpers

t.timestamps()                  # created_at + updated_at TIMESTAMP DEFAULT NOW()
t.soft_deletes()                # deleted_at TIMESTAMP NULL
t.remember_token()              # remember_token VARCHAR(100) NULL
t.polymorphic("commentable")    # commentable_id BIGINT + commentable_type VARCHAR(255) + index
t.nullable_polymorphic("taggable")

Foreign keys

t.foreign_id("role_id") \
 .constrained("roles") \
 .on_delete("cascade") \
 .on_update("cascade")

Indexes

t.unique("email", "tenant_id", name="uq_users_email_tenant")
t.index("created_at")
t.index("email", "name", name="idx_users_search")

ALTER TABLE

with Schema.table("users") as t:
    t.add_column("phone").string(20).nullable()
    t.change_column("name").string(500)
    t.drop_column("avatar")

Running in Docker

docker compose exec sdk python Program.py | flowgrate up