Skip to content

oldani/sea-query

Folders and files

NameName
Last commit message
Last commit date

Latest commit

53181b2 · Nov 5, 2024

History

95 Commits
Sep 5, 2024
Nov 5, 2024
Nov 5, 2024
Nov 5, 2024
Aug 24, 2024
Nov 5, 2024
Nov 5, 2024
Aug 24, 2024
Sep 5, 2024
Sep 20, 2024
Sep 2, 2024

Repository files navigation

SeaQuery

CI PyPI - Version PyPI - Python Version

SeaQuery is a query builder to help you construct dynamic SQL queries in Python for PostgreSQL, MySQL, and SQLite. This project is a port of the SeaQuery.rs project to Python using PyO3.

Install

You can install the package from PyPI using pip:

pip install sea-query

Or if you are using uv:

uv add sea-query

Usage

Here are some examples of how to use SeaQuery to build SQL queries:

SELECT Query

from sea_query import Query, Expr, DBEngine

query = (
    Query.select()
    .all()
    .from_table("table")
    .and_where(Expr.column("column1").ne(1))
    .and_where(Expr.column("column2").gt(2))
)
assert query.to_string(DBEngine.Postgres) == (
    'SELECT * FROM "table" WHERE "column1" <> 1 AND "column2" > 2'
)
# Or if you want to use parameter bindings
assert query.build(DBEngine.Postgres) == (
    'SELECT * FROM "table" WHERE "column1" <> $1 AND "column2" > $2',
    [1, 2]
)

INSERT Query

from sea_query import Query, Expr, DBEngine

query = (
    Query.insert()
    .into("table")
    .columns(["column1", "column2"])
    .values([1, "str1"])
    .values([2, "str2"])
)
assert query.to_string(DBEngine.Postgres) == (
    'INSERT INTO "table" ("column1", "column2") VALUES (1, \'str1\'), (2, \'str2\')'
)
# With parameter bindings
assert query.build(DBEngine.Postgres) == (
    'INSERT INTO "table" ("column1", "column2") VALUES ($1, $2), ($3, $4)',
    [1, "str1", 2, "str2"],
)

UPDATE Query

from sea_query import Query, Expr, DBEngine

query = (
    Query.update()
    .table("table")
    .value("column", 1)
    .cond_where(
        Condition.any()
        .add(Expr.column("column2").eq("value"))
        .add(Expr.column("column3").eq(3))
    )
)
assert query.to_string(DBEngine.Postgres) == (
    'UPDATE "table" SET "column" = 1 WHERE "column2" = \'value\' OR "column3" = 3'
)

DELETE Query

from sea_query import Query, Expr, DBEngine

query = (
    Query.delete()
    .from_table("table")
    .and_where(Expr.column("column1").eq(1))
)
assert query.to_string(DBEngine.Postgres) == (
    'DELETE FROM "table" WHERE "column1" = 1'
)

Table Create

from sea_query import Table, Column, DBEngine

statement = (
    Table.create()
    .name("users")
    .column(Column("id").big_integer().primary_key().auto_increment())
    .column(
        Column("name").string().string_len(128).not_null().default(Expr.value(""))
    )
    .column(Column("email").string().string_len(255).null().unique())
)

assert statement.to_string(DBEngine.Postgres) == (
    'CREATE TABLE "users" ( '
        '"id" bigserial PRIMARY KEY, '
        '"name" varchar(128) NOT NULL DEFAULT \'\', '
        '"email" varchar(255) NULL UNIQUE '
    ')'
)

Table Alter

from sea_query import Table, Column, DBEngine

statement = (
    Table.alter()
    .table("users")
    .add_column(
        Column("created_at").timestamp().null()
    )
)

assert statement.to_string(DBEngine.Postgres) == (
    'ALTER TABLE "users" ADD COLUMN "created_at" timestamp NULL'
)

Table Drop

from sea_query import Table, DBEngine

assert (
    Table.drop().table("users").to_string(DBEngine.Postgres)
    == 'DROP TABLE "users"'
)

Create Index

from sea_query import Index, DBEngine

index = (
    Index.create()
    .name("index_name")
    .   table("table")
    .column("col1")
    .column("col2")
)

assert index.to_string(DBEngine.Postgres) == (
    'CREATE INDEX "index_name" ON "table" ("col1", "col2")'
)

Drop Index

from sea_query import Index, DBEngine

index = (
    Index.drop().name("index_name").table("table")
)

assert index.to_string(DBEngine.Postgres) == (
    'DROP INDEX "index_name"'
)

About

Python binding for SeaQuery

Resources

License

Stars

Watchers

Forks

Packages

No packages published