Pure-Python PostgreSQL driver providing DB-API 2.0 compliant database connectivity with native pg8000 API and comprehensive PostgreSQL data type support.
—
Comprehensive support for all PostgreSQL data types with automatic Python conversion, including numeric types, text types, date/time types, arrays, JSON, network types, and PostgreSQL-specific types.
PostgreSQL numeric data type identifiers with their corresponding OID values.
# Integer types
BIGINT: int = 20
"""64-bit signed integer type."""
INTEGER: int = 23
"""32-bit signed integer type."""
SMALLINT: int = 21
"""16-bit signed integer type."""
# Floating point types
NUMERIC: int = 1700
"""Arbitrary precision decimal type."""
FLOAT: int = 701
"""Double precision floating point type."""
REAL: int = 700
"""Single precision floating point type."""
# Money type
MONEY: int = 790
"""Currency amount type."""
MONEY_ARRAY: int = 791
"""Array of money values."""PostgreSQL text and character data type identifiers.
TEXT: int = 25
"""Variable-length text type."""
VARCHAR: int = 1043
"""Variable-length character type with length limit."""
CHAR: int = 1042
"""Fixed-length character type."""
NAME: int = 19
"""Internal PostgreSQL name type."""
CSTRING: int = 2275
"""Null-terminated C string type."""
CSTRING_ARRAY: int = 1263
"""Array of C string values."""PostgreSQL binary data type identifiers.
BYTES: int = 17
"""Variable-length binary data type (bytea)."""
BYTES_ARRAY: int = 1001
"""Array of bytea values."""PostgreSQL temporal data type identifiers.
DATE: int = 1082
"""Date type (year, month, day)."""
TIME: int = 1083
"""Time of day type without timezone."""
TIMESTAMP: int = 1114
"""Date and time type without timezone."""
TIMESTAMPTZ: int = 1184
"""Date and time type with timezone."""
INTERVAL: int = 1186
"""Time interval type."""
TIME_ARRAY: int = 1183
"""Array of time values."""
INTERVAL_ARRAY: int = 1187
"""Array of interval values."""PostgreSQL boolean data type identifier.
BOOLEAN: int = 16
"""Boolean true/false type."""PostgreSQL JSON data type identifiers.
JSON: int = 114
"""JSON data type (text-based storage)."""
JSONB: int = 3802
"""JSONB data type (binary storage with indexing)."""
JSON_ARRAY: int = 199
"""Array of JSON values."""
JSONB_ARRAY: int = 3807
"""Array of JSONB values."""PostgreSQL network address data type identifiers.
INET: int = 869
"""IPv4 or IPv6 network address type."""
CIDR: int = 650
"""IPv4 or IPv6 network specification type."""
MACADDR: int = 829
"""MAC address type."""
CIDR_ARRAY: int = 651
"""Array of CIDR values."""
INET_ARRAY: int = 1041
"""Array of INET values."""PostgreSQL range data type identifiers for various base types.
INT4RANGE: int = 3904
"""Range of integer values."""
INT8RANGE: int = 3926
"""Range of bigint values."""
NUMRANGE: int = 3906
"""Range of numeric values."""
TSRANGE: int = 3908
"""Range of timestamp values."""
TSTZRANGE: int = 3910
"""Range of timestamp with timezone values."""
DATERANGE: int = 3912
"""Range of date values."""
# Range Arrays
INT4RANGE_ARRAY: int = 3905
"""Array of int4range values."""
INT8RANGE_ARRAY: int = 3927
"""Array of int8range values."""
NUMRANGE_ARRAY: int = 3907
"""Array of numrange values."""
TSRANGE_ARRAY: int = 3909
"""Array of tsrange values."""
TSTZRANGE_ARRAY: int = 3911
"""Array of tstzrange values."""
DATERANGE_ARRAY: int = 3913
"""Array of daterange values."""
# Multirange Types (PostgreSQL 14+)
INT4MULTIRANGE: int = 4451
"""Multirange of integer values."""
INT8MULTIRANGE: int = 4536
"""Multirange of bigint values."""
NUMMULTIRANGE: int = 4532
"""Multirange of numeric values."""
TSMULTIRANGE: int = 4533
"""Multirange of timestamp values."""
TSTZMULTIRANGE: int = 4534
"""Multirange of timestamp with timezone values."""
DATEMULTIRANGE: int = 4535
"""Multirange of date values."""
# Multirange Arrays
INT4MULTIRANGE_ARRAY: int = 6150
"""Array of int4multirange values."""
INT8MULTIRANGE_ARRAY: int = 6157
"""Array of int8multirange values."""
NUMMULTIRANGE_ARRAY: int = 6151
"""Array of nummultirange values."""
TSMULTIRANGE_ARRAY: int = 6152
"""Array of tsmultirange values."""
TSTZMULTIRANGE_ARRAY: int = 6153
"""Array of tstzmultirange values."""
DATEMULTIRANGE_ARRAY: int = 6155
"""Array of datemultirange values."""Array variants for all base PostgreSQL data types.
# Numeric arrays
INTEGER_ARRAY: int = 1007
"""Array of integer values."""
BIGINT_ARRAY: int = 1016
"""Array of bigint values."""
SMALLINT_ARRAY: int = 1005
"""Array of smallint values."""
FLOAT_ARRAY: int = 1022
"""Array of double precision float values."""
REAL_ARRAY: int = 1021
"""Array of single precision float values."""
NUMERIC_ARRAY: int = 1231
"""Array of numeric values."""
# Text arrays
TEXT_ARRAY: int = 1009
"""Array of text values."""
VARCHAR_ARRAY: int = 1015
"""Array of varchar values."""
CHAR_ARRAY: int = 1014
"""Array of char values."""
NAME_ARRAY: int = 1003
"""Array of name values."""
# Other arrays
BOOLEAN_ARRAY: int = 1000
"""Array of boolean values."""
DATE_ARRAY: int = 1182
"""Array of date values."""
TIMESTAMP_ARRAY: int = 1115
"""Array of timestamp values."""
TIMESTAMPTZ_ARRAY: int = 1185
"""Array of timestamp with timezone values."""Additional PostgreSQL data type identifiers.
UUID_TYPE: int = 2950
"""Universally unique identifier type."""
UUID_ARRAY: int = 2951
"""Array of UUID values."""
OID: int = 26
"""Object identifier type."""
XID: int = 28
"""Transaction identifier type."""
POINT: int = 600
"""Geometric point type."""
UNKNOWN: int = 705
"""Unknown type placeholder."""
NULLTYPE: int = -1
"""Null type identifier."""
INT2VECTOR: int = 22
"""Vector of 16-bit integers type."""
RECORD: int = 2249
"""Anonymous record type."""
ANY_ARRAY: int = 2277
"""Pseudo-type representing any array."""Functions for converting PostgreSQL data to Python objects.
def bool_in(data: str) -> bool:
"""Convert PostgreSQL boolean string to Python bool."""
def bytes_in(data: str) -> bytes:
"""Convert PostgreSQL bytea hex string to Python bytes."""
def cidr_in(data: str) -> ipaddress.IPv4Network | ipaddress.IPv6Network | ipaddress.IPv4Address | ipaddress.IPv6Address:
"""Convert PostgreSQL CIDR/INET string to Python IP object."""
def date_in(data: str) -> datetime.date | str:
"""Convert PostgreSQL date string to Python date (or string for infinity)."""
def inet_in(data: str) -> ipaddress.IPv4Network | ipaddress.IPv6Network | ipaddress.IPv4Address | ipaddress.IPv6Address:
"""Convert PostgreSQL INET string to Python IP object."""
def int_in(data: str) -> int:
"""Convert PostgreSQL integer string to Python int."""
def interval_in(data: str) -> datetime.timedelta | PGInterval:
"""Convert PostgreSQL interval string to Python timedelta or PGInterval."""
def json_in(data: str) -> object:
"""Convert PostgreSQL JSON string to Python object."""
def numeric_in(data: str) -> decimal.Decimal:
"""Convert PostgreSQL numeric string to Python Decimal."""
def point_in(data: str) -> tuple:
"""Convert PostgreSQL point string to Python tuple (x, y)."""
def string_in(data: str) -> str:
"""Pass-through for string data."""
def time_in(data: str) -> datetime.time:
"""Convert PostgreSQL time string to Python time."""
def timestamp_in(data: str) -> datetime.datetime | str:
"""Convert PostgreSQL timestamp string to Python datetime (or string for infinity)."""
def timestamptz_in(data: str) -> datetime.datetime | str:
"""Convert PostgreSQL timestamptz string to Python datetime with timezone."""
def uuid_in(data: str) -> uuid.UUID:
"""Convert PostgreSQL UUID string to Python UUID."""
def vector_in(data: str) -> list:
"""Convert PostgreSQL int vector to Python list."""
def record_in(data: str) -> tuple:
"""Convert PostgreSQL record/tuple string to Python tuple."""Functions for converting Python objects to PostgreSQL data.
def bool_out(v: bool) -> str:
"""Convert Python bool to PostgreSQL boolean string."""
def bytes_out(v: bytes) -> str:
"""Convert Python bytes to PostgreSQL bytea hex string."""
def cidr_out(v) -> str:
"""Convert Python IP object to PostgreSQL CIDR string."""
def date_out(v: datetime.date) -> str:
"""Convert Python date to PostgreSQL date string."""
def datetime_out(v: datetime.datetime) -> str:
"""Convert Python datetime to PostgreSQL timestamp string."""
def enum_out(v: Enum) -> str:
"""Convert Python enum to PostgreSQL enum string."""
def float_out(v: float) -> str:
"""Convert Python float to PostgreSQL float string."""
def inet_out(v) -> str:
"""Convert Python IP object to PostgreSQL INET string."""
def int_out(v: int) -> str:
"""Convert Python int to PostgreSQL integer string."""
def interval_out(v: datetime.timedelta) -> str:
"""Convert Python timedelta to PostgreSQL interval string."""
def json_out(v: object) -> str:
"""Convert Python object to PostgreSQL JSON string."""
def numeric_out(d: decimal.Decimal) -> str:
"""Convert Python Decimal to PostgreSQL numeric string."""
def pg_interval_out(v: PGInterval) -> str:
"""Convert PGInterval to PostgreSQL interval string."""
def range_out(v: Range) -> str:
"""Convert Python Range to PostgreSQL range string."""
def string_out(v: str) -> str:
"""Pass-through for string data."""
def time_out(v: datetime.time) -> str:
"""Convert Python time to PostgreSQL time string."""
def uuid_out(v: uuid.UUID) -> str:
"""Convert Python UUID to PostgreSQL UUID string."""Functions for converting PostgreSQL range and multirange types.
# Range Input Functions
def daterange_in(data: str) -> Range:
"""Convert PostgreSQL daterange to Python Range with date bounds."""
def int4range_in(data: str) -> Range:
"""Convert PostgreSQL int4range to Python Range with integer bounds."""
def int8range_in(data: str) -> Range:
"""Convert PostgreSQL int8range to Python Range with integer bounds."""
def numrange_in(data: str) -> Range:
"""Convert PostgreSQL numrange to Python Range with Decimal bounds."""
def tsrange_in(data: str) -> Range:
"""Convert PostgreSQL tsrange to Python Range with timestamp bounds."""
def tstzrange_in(data: str) -> Range:
"""Convert PostgreSQL tstzrange to Python Range with timestamptz bounds."""
# Multirange Input Functions
def datemultirange_in(data: str) -> list:
"""Convert PostgreSQL datemultirange to list of date ranges."""
def int4multirange_in(data: str) -> list:
"""Convert PostgreSQL int4multirange to list of integer ranges."""
def int8multirange_in(data: str) -> list:
"""Convert PostgreSQL int8multirange to list of integer ranges."""
def nummultirange_in(data: str) -> list:
"""Convert PostgreSQL nummultirange to list of numeric ranges."""
def tsmultirange_in(data: str) -> list:
"""Convert PostgreSQL tsmultirange to list of timestamp ranges."""
def tstzmultirange_in(data: str) -> list:
"""Convert PostgreSQL tstzmultirange to list of timestamptz ranges."""Functions for converting PostgreSQL array types to Python lists.
def bool_array_in(data: str) -> list:
"""Convert PostgreSQL bool[] to Python list of bools."""
def bytes_array_in(data: str) -> list:
"""Convert PostgreSQL bytea[] to Python list of bytes."""
def date_array_in(data: str) -> list:
"""Convert PostgreSQL date[] to Python list of dates."""
def float_array_in(data: str) -> list:
"""Convert PostgreSQL float[] to Python list of floats."""
def int_array_in(data: str) -> list:
"""Convert PostgreSQL int[] to Python list of integers."""
def json_array_in(data: str) -> list:
"""Convert PostgreSQL json[] to Python list of JSON objects."""
def numeric_array_in(data: str) -> list:
"""Convert PostgreSQL numeric[] to Python list of Decimals."""
def string_array_in(data: str) -> list:
"""Convert PostgreSQL text[] to Python list of strings."""
def timestamp_array_in(data: str) -> list:
"""Convert PostgreSQL timestamp[] to Python list of timestamps."""
def uuid_array_in(data: str) -> list:
"""Convert PostgreSQL uuid[] to Python list of UUIDs."""
# Additional specialized array converters available for all supported typesHelper functions for SQL formatting and parameter conversion.
def identifier(sql: str) -> str:
"""Escape SQL identifier for safe use in queries."""
@singledispatch
def literal(value: object) -> str:
"""
Create SQL literal representation of Python value.
Supports type-specific formatting via singledispatch for:
- None: Returns "NULL"
- bool: Returns "TRUE" or "FALSE"
- int/float/Decimal: Returns string representation
- bytes/bytearray: Returns hex format with E'' wrapper
- datetime types: Returns quoted strings
- list: Returns quoted array string
- str: Returns escaped and quoted string
"""
@singledispatch
def array_out(val: object) -> str:
"""
Convert Python values to PostgreSQL array format.
Supports type-specific formatting via singledispatch for:
- list: Converts to PostgreSQL array string
- tuple: Converts to PostgreSQL composite string
- None: Returns "NULL"
- dict: Converts to escaped JSON string
- bytes/bytearray: Converts to quoted hex string
- str: Converts with proper escaping
"""
@singledispatch
def composite_out(val: object) -> str:
"""Convert Python values to PostgreSQL composite format."""
def array_string_escape(v: str) -> str:
"""Escape strings for PostgreSQL array format."""
def make_param(py_types: dict, value: object) -> str:
"""Convert Python value using appropriate converter."""
def make_params(py_types: dict, values) -> tuple:
"""Convert multiple Python values using appropriate converters."""Specialized functions for PostgreSQL interval and time handling.
def pginterval_in(data: bytes) -> PGInterval:
"""
Parse PostgreSQL interval data into PGInterval object.
Parameters:
- data: Raw PostgreSQL interval data
Returns:
PGInterval object representing the interval
"""
def pginterval_out(v: PGInterval) -> bytes:
"""
Format PGInterval object for PostgreSQL storage.
Parameters:
- v: PGInterval object to format
Returns:
Formatted interval data for PostgreSQL
"""
def timedelta_in(data: bytes) -> datetime.timedelta:
"""
Convert PostgreSQL interval to Python timedelta.
Parameters:
- data: Raw PostgreSQL interval data
Returns:
Python timedelta object
"""Legacy type name aliases for backward compatibility.
# Legacy numeric aliases
BIGINTEGER: int = BIGINT
"""Alias for BIGINT type."""
DATETIME: int = TIMESTAMP
"""Alias for TIMESTAMP type."""
NUMBER: int = NUMERIC
"""Alias for NUMERIC type."""
DECIMAL: int = NUMERIC
"""Alias for NUMERIC type."""
DECIMAL_ARRAY: int = NUMERIC_ARRAY
"""Alias for NUMERIC_ARRAY type."""
ROWID: int = OID
"""Alias for OID type."""
TIMEDELTA: int = INTERVAL
"""Alias for INTERVAL type."""
STRING: int = VARCHAR
"""Alias for VARCHAR type."""import pg8000
import datetime
import decimal
import uuid
import ipaddress
conn = pg8000.connect(user="myuser", password="mypass", database="mydb")
# Insert various data types
cursor = conn.cursor()
cursor.execute("""
INSERT INTO test_types (
int_col, bigint_col, numeric_col, float_col,
text_col, bool_col, date_col, timestamp_col,
json_col, uuid_col, inet_col, bytea_col
) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)
""", (
42, # INTEGER
9223372036854775807, # BIGINT
decimal.Decimal('123.456'), # NUMERIC
3.14159, # FLOAT
"Hello, PostgreSQL!", # TEXT
True, # BOOLEAN
datetime.date(2023, 12, 25), # DATE
datetime.datetime.now(), # TIMESTAMP
{"key": "value", "array": [1,2,3]}, # JSON
uuid.uuid4(), # UUID
ipaddress.IPv4Address('192.168.1.1'), # INET
b"binary data" # BYTEA
))
conn.commit()
cursor.close()
conn.close()import pg8000
conn = pg8000.connect(user="myuser", password="mypass", database="mydb")
cursor = conn.cursor()
# Insert array data
cursor.execute("""
INSERT INTO test_arrays (
int_array, text_array, bool_array
) VALUES (%s, %s, %s)
""", (
[1, 2, 3, 4, 5], # INTEGER_ARRAY
["apple", "banana", "cherry"], # TEXT_ARRAY
[True, False, True, False] # BOOLEAN_ARRAY
))
# Query array data
cursor.execute("SELECT int_array, text_array FROM test_arrays WHERE id = %s", (1,))
row = cursor.fetchone()
print(f"Integer array: {row[0]}")
print(f"Text array: {row[1]}")
conn.commit()
cursor.close()
conn.close()import pg8000
import json
conn = pg8000.connect(user="myuser", password="mypass", database="mydb")
cursor = conn.cursor()
# Insert JSON data
user_profile = {
"name": "John Doe",
"age": 30,
"interests": ["programming", "music", "travel"],
"address": {
"street": "123 Main St",
"city": "Anytown",
"zip": "12345"
}
}
cursor.execute("""
INSERT INTO user_profiles (user_id, profile_data)
VALUES (%s, %s)
""", (123, user_profile))
# Query JSON data with JSON operators
cursor.execute("""
SELECT profile_data->>'name' as name,
profile_data->'interests' as interests
FROM user_profiles
WHERE user_id = %s
""", (123,))
row = cursor.fetchone()
print(f"Name: {row[0]}")
print(f"Interests: {row[1]}")
conn.commit()
cursor.close()
conn.close()Install with Tessl CLI
npx tessl i tessl/pypi-pg8000