pg

Postgres-specific column types.

BigInt

Bases: Column[int]

Big integer column type.

Source code in src/embar/column/pg.py

class BigInt(Column[int]):
    """
    Big integer column type.
    """

    _sql_type: str = "BIGINT"
    _py_type: Type = int

BigSerial

Bases: Column[int]

Auto-incrementing big integer column.

Source code in src/embar/column/pg.py

class BigSerial(Column[int]):
    """
    Auto-incrementing big integer column.
    """

    _sql_type: str = "BIGSERIAL"
    _py_type: Type = int

Boolean

Bases: Column[bool]

Boolean column type.

Source code in src/embar/column/pg.py

class Boolean(Column[bool]):
    """
    Boolean column type.
    """

    _sql_type: str = "BOOLEAN"
    _py_type: Type = bool

Char

Bases: Column[str]

Fixed-length character column type.

Source code in src/embar/column/pg.py

class Char(Column[str]):
    """
    Fixed-length character column type.
    """

    _sql_type: str = "CHAR"
    _py_type: Type = str

    def __init__(
        self,
        name: str | None = None,
        default: str | None = None,
        primary: bool = False,
        not_null: bool = False,
        length: int | None = None,
    ):
        """
        Create a new Char instance.
        """
        self._extra_args: tuple[int] | tuple[int, int] | None = (length,) if length is not None else None
        super().__init__(name=name, default=default, primary=primary, not_null=not_null)

__init__(name=None, default=None, primary=False, not_null=False, length=None)

Create a new Char instance.

Source code in src/embar/column/pg.py

def __init__(
    self,
    name: str | None = None,
    default: str | None = None,
    primary: bool = False,
    not_null: bool = False,
    length: int | None = None,
):
    """
    Create a new Char instance.
    """
    self._extra_args: tuple[int] | tuple[int, int] | None = (length,) if length is not None else None
    super().__init__(name=name, default=default, primary=primary, not_null=not_null)

Date

Bases: Column[date]

Date column type.

Source code in src/embar/column/pg.py

class Date(Column[date]):
    """
    Date column type.
    """

    _sql_type: str = "DATE"
    _py_type: Type = date

DoublePrecision

Bases: Column[float]

Double precision floating point column type.

Source code in src/embar/column/pg.py

class DoublePrecision(Column[float]):
    """
    Double precision floating point column type.
    """

    _sql_type: str = "DOUBLE PRECISION"
    _py_type: Type = float

EmbarEnum

Bases: str, Enum

EmbarEnum is just a regular Enum but without having to set the right side.

from enum import auto
from embar.column.pg import EmbarEnum
class StatusEnum(EmbarEnum):
   PENDING = auto()
   DONE = auto()

Source code in src/embar/column/pg.py

class EmbarEnum(str, Enum):
    """
    `EmbarEnum` is just a regular Enum but without having to set the right side.

    ```python
    from enum import auto
    from embar.column.pg import EmbarEnum
    class StatusEnum(EmbarEnum):
       PENDING = auto()
       DONE = auto()
    ```
    """

    @staticmethod
    @override
    def _generate_next_value_(name: str, start: int, count: int, last_values: list[Any]) -> str:
        return name

EnumCol

Bases: Column[str]

Column type for Postgres enum values.

Source code in src/embar/column/pg.py

class EnumCol[E: EmbarEnum](Column[str]):
    """
    Column type for Postgres enum values.
    """

    _sql_type: str
    _py_type: Type = str

    def __init__(
        self,
        pg_enum: type[PgEnum[E]],
        name: str | None = None,
        default: E | None = None,
        primary: bool = False,
        not_null: bool = False,
    ):
        """
        Create a new EnumCol instance.
        """
        self._sql_type = pg_enum.name

        super().__init__(name=name, default=default, primary=primary, not_null=not_null)

__init__(pg_enum, name=None, default=None, primary=False, not_null=False)

Create a new EnumCol instance.

Source code in src/embar/column/pg.py

def __init__(
    self,
    pg_enum: type[PgEnum[E]],
    name: str | None = None,
    default: E | None = None,
    primary: bool = False,
    not_null: bool = False,
):
    """
    Create a new EnumCol instance.
    """
    self._sql_type = pg_enum.name

    super().__init__(name=name, default=default, primary=primary, not_null=not_null)

Float

Bases: Column[float]

A floating point column type.

Source code in src/embar/column/common.py

class Float(Column[float]):
    """
    A floating point column type.
    """

    _sql_type: str = "REAL"
    _py_type: Type = float

Integer

Bases: Column[int]

An integer column type.

Source code in src/embar/column/common.py

class Integer(Column[int]):
    """
    An integer column type.
    """

    _sql_type: str = "INTEGER"
    _py_type: Type = int

Interval

Bases: Column[timedelta]

Interval column type for storing time intervals.

Source code in src/embar/column/pg.py

class Interval(Column[timedelta]):
    """
    Interval column type for storing time intervals.
    """

    _sql_type: str = "INTERVAL"
    _py_type: Type = timedelta

Json

Bases: Column[dict[str, Any]]

JSON column type for storing JSON data.

Source code in src/embar/column/pg.py

class Json(Column[dict[str, Any]]):
    """
    JSON column type for storing JSON data.
    """

    _sql_type: str = "JSON"
    _py_type: Type = dict[str, Any]

Jsonb

Bases: Column[dict[str, Any]]

JSONB column type for storing JSON data.

Source code in src/embar/column/pg.py

class Jsonb(Column[dict[str, Any]]):
    """
    JSONB column type for storing JSON data.
    """

    _sql_type: str = "JSONB"
    _py_type: Type = dict[str, Any]

Numeric

Bases: Column[Decimal]

Numeric column type with configurable precision and scale.

Source code in src/embar/column/pg.py

class Numeric(Column[Decimal]):
    """
    Numeric column type with configurable precision and scale.
    """

    _sql_type: str = "NUMERIC"
    _py_type: Type = Decimal

    _extra_args: tuple[int] | tuple[int, int] | None

    def __init__(
        self,
        name: str | None = None,
        default: Decimal | None = None,
        primary: bool = False,
        not_null: bool = False,
        precision: int | None = None,
        scale: int | None = None,
    ):
        """
        Create a new Numeric instance.
        """
        if precision is None:
            if scale is not None:
                raise Exception("Numeric: 'precision' cannot be None if scale is set")
        elif scale is None:
            self._extra_args = (precision,)
        else:
            self._extra_args = (precision, scale)
        super().__init__(name=name, default=default, primary=primary, not_null=not_null)

__init__(name=None, default=None, primary=False, not_null=False, precision=None, scale=None)

Create a new Numeric instance.

Source code in src/embar/column/pg.py

def __init__(
    self,
    name: str | None = None,
    default: Decimal | None = None,
    primary: bool = False,
    not_null: bool = False,
    precision: int | None = None,
    scale: int | None = None,
):
    """
    Create a new Numeric instance.
    """
    if precision is None:
        if scale is not None:
            raise Exception("Numeric: 'precision' cannot be None if scale is set")
    elif scale is None:
        self._extra_args = (precision,)
    else:
        self._extra_args = (precision, scale)
    super().__init__(name=name, default=default, primary=primary, not_null=not_null)

PgDecimal

Bases: Column[Decimal]

Decimal column type with configurable precision and scale.

Source code in src/embar/column/pg.py

class PgDecimal(Column[Decimal]):
    """
    Decimal column type with configurable precision and scale.
    """

    # Note: DECIMAL is an alias for NUMERIC in PostgreSQL
    _sql_type: str = "DECIMAL"
    _py_type: Type = Decimal

    _extra_args: tuple[int] | tuple[int, int] | None

    def __init__(
        self,
        name: str | None = None,
        default: Decimal | None = None,
        primary: bool = False,
        not_null: bool = False,
        precision: int | None = None,
        scale: int | None = None,
    ):
        """
        Create a new PgDecimal instance.
        """
        if precision is None:
            if scale is not None:
                raise Exception("Numeric: 'precision' cannot be None if scale is set")
        elif scale is None:
            self._extra_args = (precision,)
        else:
            self._extra_args = (precision, scale)
        super().__init__(name=name, default=default, primary=primary, not_null=not_null)

__init__(name=None, default=None, primary=False, not_null=False, precision=None, scale=None)

Create a new PgDecimal instance.

Source code in src/embar/column/pg.py

def __init__(
    self,
    name: str | None = None,
    default: Decimal | None = None,
    primary: bool = False,
    not_null: bool = False,
    precision: int | None = None,
    scale: int | None = None,
):
    """
    Create a new PgDecimal instance.
    """
    if precision is None:
        if scale is not None:
            raise Exception("Numeric: 'precision' cannot be None if scale is set")
    elif scale is None:
        self._extra_args = (precision,)
    else:
        self._extra_args = (precision, scale)
    super().__init__(name=name, default=default, primary=primary, not_null=not_null)

PgEnum

Bases: EnumBase

`PgEnum is used to create Postgres enum types.

Subclasses must always assign values to the two class variables!

from enum import auto
from embar.table import Table
from embar.column.pg import EmbarEnum, EnumCol, PgEnum
class StatusEnum(EmbarEnum):
   PENDING = auto()
   DONE = auto()
class StatusPgEnum(PgEnum[StatusEnum]):
    name: str = "status_enum"
    enum: type[StatusEnum] = StatusEnum
class TableWithStatus(Table):
    status: EnumCol[StatusEnum] = EnumCol(StatusPgEnum)

Source code in src/embar/column/pg.py

class PgEnum[E: EmbarEnum](EnumBase):
    """
    `PgEnum is used to create Postgres enum types.

    Subclasses must always assign values to the two class variables!

    ```python
    from enum import auto
    from embar.table import Table
    from embar.column.pg import EmbarEnum, EnumCol, PgEnum
    class StatusEnum(EmbarEnum):
       PENDING = auto()
       DONE = auto()
    class StatusPgEnum(PgEnum[StatusEnum]):
        name: str = "status_enum"
        enum: type[StatusEnum] = StatusEnum
    class TableWithStatus(Table):
        status: EnumCol[StatusEnum] = EnumCol(StatusPgEnum)
    ```
    """

    name: str
    enum: type[E]

    @override
    @classmethod
    def ddl(cls) -> str:
        quoted = [f"'{e.name}'" for e in cls.enum]
        values = ", ".join(quoted)
        sql = f"CREATE TYPE {cls.name} AS ENUM ({values});"
        return sql

Serial

Bases: Column[int]

Auto-incrementing integer column.

Source code in src/embar/column/pg.py

class Serial(Column[int]):
    """
    Auto-incrementing integer column.
    """

    _sql_type: str = "SERIAL"
    _py_type: Type = int

SmallInt

Bases: Column[int]

Small integer column type.

Source code in src/embar/column/pg.py

class SmallInt(Column[int]):
    """
    Small integer column type.
    """

    _sql_type: str = "SMALLINT"
    _py_type: Type = int

SmallSerial

Bases: Column[int]

Auto-incrementing small integer column.

Source code in src/embar/column/pg.py

class SmallSerial(Column[int]):
    """
    Auto-incrementing small integer column.
    """

    _sql_type: str = "SMALLSERIAL"
    _py_type: Type = int

Text

Bases: Column[str]

A text column type.

Source code in src/embar/column/common.py

class Text(Column[str]):
    """
    A text column type.
    """

    _sql_type: str = "TEXT"
    _py_type: Type = str

Time

Bases: Column[time]

Time column type.

Source code in src/embar/column/pg.py

class Time(Column[time]):
    """
    Time column type.
    """

    _sql_type: str = "TIME"
    _py_type: Type = time

Timestamp

Bases: Column[datetime]

Timestamp column type.

Source code in src/embar/column/pg.py

class Timestamp(Column[datetime]):
    """
    Timestamp column type.
    """

    _sql_type: str = "TIMESTAMP"
    _py_type: Type = str

Varchar

Bases: Column[str]

Variable-length character column type.

Source code in src/embar/column/pg.py

class Varchar(Column[str]):
    """
    Variable-length character column type.
    """

    _sql_type: str = "VARCHAR"
    _py_type: Type = str

    def __init__(
        self,
        name: str | None = None,
        default: str | None = None,
        primary: bool = False,
        not_null: bool = False,
        length: int | None = None,
    ):
        """
        Create a new Varchar instance.
        """
        self._extra_args: tuple[int] | tuple[int, int] | None = (length,) if length is not None else None
        super().__init__(name=name, default=default, primary=primary, not_null=not_null)

__init__(name=None, default=None, primary=False, not_null=False, length=None)

Create a new Varchar instance.

Source code in src/embar/column/pg.py

def __init__(
    self,
    name: str | None = None,
    default: str | None = None,
    primary: bool = False,
    not_null: bool = False,
    length: int | None = None,
):
    """
    Create a new Varchar instance.
    """
    self._extra_args: tuple[int] | tuple[int, int] | None = (length,) if length is not None else None
    super().__init__(name=name, default=default, primary=primary, not_null=not_null)

Vector

Bases: Column[list[float]]

Vector column using pgvector.

This assumes the extension is already installed and activated with CREATE EXTENSION vector;

Source code in src/embar/column/pg.py

class Vector(Column[list[float]]):
    """
    Vector column using [pgvector](https://github.com/pgvector/pgvector).

    This assumes the extension is already installed and activated with
    CREATE EXTENSION vector;
    """

    _sql_type: str = "VECTOR"
    _py_type: Type = list[float]

    def __init__(
        self,
        length: int,
        name: str | None = None,
        default: list[float] | None = None,
        primary: bool = False,
        not_null: bool = False,
    ):
        """
        Create a new Vector instance.
        """
        self._extra_args: tuple[int] | tuple[int, int] | None = (length,)
        super().__init__(name=name, default=default, primary=primary, not_null=not_null)

__init__(length, name=None, default=None, primary=False, not_null=False)

Create a new Vector instance.

Source code in src/embar/column/pg.py

def __init__(
    self,
    length: int,
    name: str | None = None,
    default: list[float] | None = None,
    primary: bool = False,
    not_null: bool = False,
):
    """
    Create a new Vector instance.
    """
    self._extra_args: tuple[int] | tuple[int, int] | None = (length,)
    super().__init__(name=name, default=default, primary=primary, not_null=not_null)