Skip to content

graflo.db.postgres.types

Type mapping utilities for PostgreSQL to graflo type conversion.

This module provides utilities for mapping PostgreSQL data types to graflo Field types, enabling automatic schema inference from PostgreSQL database schemas.

PostgresTypeMapper

Maps PostgreSQL data types to graflo Field types.

This class provides static methods for converting PostgreSQL type names (from information_schema or pg_catalog) to graflo Field type strings.

Source code in graflo/db/postgres/types.py 
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
class PostgresTypeMapper:
    """Maps PostgreSQL data types to graflo Field types.

    This class provides static methods for converting PostgreSQL type names
    (from information_schema or pg_catalog) to graflo Field type strings.
    """

    # Mapping of PostgreSQL types to graflo Field types
    TYPE_MAPPING = {
        # Integer types
        "integer": "INT",
        "int": "INT",
        "int4": "INT",
        "smallint": "INT",
        "int2": "INT",
        "bigint": "INT",
        "int8": "INT",
        "serial": "INT",
        "bigserial": "INT",
        "smallserial": "INT",
        # Floating point types
        "real": "FLOAT",
        "float4": "FLOAT",
        "double precision": "FLOAT",
        "float8": "FLOAT",
        "numeric": "FLOAT",
        "decimal": "FLOAT",
        # Boolean
        "boolean": "BOOL",
        "bool": "BOOL",
        # String types
        "character varying": "STRING",
        "varchar": "STRING",
        "character": "STRING",
        "char": "STRING",
        "text": "STRING",
        # Date/time types (mapped to DATETIME)
        "timestamp": "DATETIME",
        "timestamp without time zone": "DATETIME",
        "timestamp with time zone": "DATETIME",
        "timestamptz": "DATETIME",
        "date": "DATETIME",
        "time": "DATETIME",
        "time without time zone": "DATETIME",
        "time with time zone": "DATETIME",
        "timetz": "DATETIME",
        "interval": "STRING",  # Interval is duration, keep as STRING
        # JSON types
        "json": "STRING",
        "jsonb": "STRING",
        # Binary types
        "bytea": "STRING",
        # UUID
        "uuid": "STRING",
    }

    @classmethod
    def map_type(cls, postgres_type: str) -> str:
        """Map PostgreSQL type to graflo Field type.

        Array types (``integer[]``, ``text[]``, …) map to ``LIST``; use
        :meth:`map_field` when ``item_type`` is needed.

        Args:
            postgres_type: PostgreSQL type name (e.g., 'int4', 'varchar', 'timestamp')

        Returns:
            str: graflo Field type (INT, FLOAT, BOOL, STRING, LIST, …)
        """
        field_type, _item_type = cls.map_field(postgres_type)
        return field_type

    @classmethod
    def map_field(cls, postgres_type: str) -> tuple[str, str | None]:
        """Map PostgreSQL type to ``(FieldType, item_type|None)``.

        Homogeneous SQL arrays become ``("LIST", <scalar>)`` when the element
        type is known; otherwise ``("LIST", None)`` is avoided — unknown element
        types fall through to STRING rather than inventing a wrong item_type.
        """
        normalized = postgres_type.lower().strip()

        if "(" in normalized:
            normalized = normalized.split("(")[0].strip()

        is_array = False
        if normalized.endswith("[]"):
            is_array = True
            normalized = normalized[:-2].strip()
        elif normalized.startswith("_") and normalized[1:] in cls.TYPE_MAPPING:
            # pg catalog array aliases like _int4, _text
            is_array = True
            normalized = normalized[1:]

        mapped: str | None = None
        if normalized in cls.TYPE_MAPPING:
            mapped = cls.TYPE_MAPPING[normalized]
        else:
            for pg_type, graflo_type in cls.TYPE_MAPPING.items():
                if pg_type in normalized or normalized in pg_type:
                    logger.debug(
                        f"Mapped PostgreSQL type '{postgres_type}' to graflo type "
                        f"'{graflo_type}' (partial match with '{pg_type}')"
                    )
                    mapped = graflo_type
                    break

        if mapped is None:
            logger.warning(
                f"Unknown PostgreSQL type '{postgres_type}', defaulting to STRING"
            )
            mapped = "STRING"

        if is_array:
            if mapped in {
                "INT",
                "FLOAT",
                "BOOL",
                "STRING",
                "DATETIME",
            }:
                return "LIST", mapped
            # Do not invent a wrong scalar item_type
            logger.debug(
                "PostgreSQL array type '%s' mapped without reliable item_type",
                postgres_type,
            )
            return "LIST", None

        return mapped, None

    @classmethod
    def is_datetime_type(cls, postgres_type: str) -> bool:
        """Check if a PostgreSQL type is a datetime type.

        Args:
            postgres_type: PostgreSQL type name

        Returns:
            bool: True if the type is a datetime-related type
        """
        normalized = postgres_type.lower().strip()
        datetime_types = [
            "timestamp",
            "date",
            "time",
            "interval",
            "timestamptz",
            "timetz",
        ]
        return any(dt_type in normalized for dt_type in datetime_types)

    @classmethod
    def is_numeric_type(cls, postgres_type: str) -> bool:
        """Check if a PostgreSQL type is a numeric type.

        Args:
            postgres_type: PostgreSQL type name

        Returns:
            bool: True if the type is numeric
        """
        normalized = postgres_type.lower().strip()
        numeric_types = [
            "integer",
            "int",
            "bigint",
            "smallint",
            "serial",
            "real",
            "double precision",
            "numeric",
            "decimal",
            "float",
        ]
        return any(nt_type in normalized for nt_type in numeric_types)

is_datetime_type(postgres_type) classmethod

Check if a PostgreSQL type is a datetime type.

Parameters:

Name Type Description Default
postgres_type str

PostgreSQL type name

 required

Returns:

Name Type Description
bool bool

True if the type is a datetime-related type
Source code in graflo/db/postgres/types.py 
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
@classmethod
def is_datetime_type(cls, postgres_type: str) -> bool:
    """Check if a PostgreSQL type is a datetime type.

    Args:
        postgres_type: PostgreSQL type name

    Returns:
        bool: True if the type is a datetime-related type
    """
    normalized = postgres_type.lower().strip()
    datetime_types = [
        "timestamp",
        "date",
        "time",
        "interval",
        "timestamptz",
        "timetz",
    ]
    return any(dt_type in normalized for dt_type in datetime_types)

is_numeric_type(postgres_type) classmethod

Check if a PostgreSQL type is a numeric type.

Parameters:

Name Type Description Default
postgres_type str

PostgreSQL type name

 required

Returns:

Name Type Description
bool bool

True if the type is numeric
Source code in graflo/db/postgres/types.py 
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
@classmethod
def is_numeric_type(cls, postgres_type: str) -> bool:
    """Check if a PostgreSQL type is a numeric type.

    Args:
        postgres_type: PostgreSQL type name

    Returns:
        bool: True if the type is numeric
    """
    normalized = postgres_type.lower().strip()
    numeric_types = [
        "integer",
        "int",
        "bigint",
        "smallint",
        "serial",
        "real",
        "double precision",
        "numeric",
        "decimal",
        "float",
    ]
    return any(nt_type in normalized for nt_type in numeric_types)

map_field(postgres_type) classmethod

Map PostgreSQL type to (FieldType, item_type|None).

Homogeneous SQL arrays become ("LIST", <scalar>) when the element type is known; otherwise ("LIST", None) is avoided — unknown element types fall through to STRING rather than inventing a wrong item_type.

Source code in graflo/db/postgres/types.py 
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
@classmethod
def map_field(cls, postgres_type: str) -> tuple[str, str | None]:
    """Map PostgreSQL type to ``(FieldType, item_type|None)``.

    Homogeneous SQL arrays become ``("LIST", <scalar>)`` when the element
    type is known; otherwise ``("LIST", None)`` is avoided — unknown element
    types fall through to STRING rather than inventing a wrong item_type.
    """
    normalized = postgres_type.lower().strip()

    if "(" in normalized:
        normalized = normalized.split("(")[0].strip()

    is_array = False
    if normalized.endswith("[]"):
        is_array = True
        normalized = normalized[:-2].strip()
    elif normalized.startswith("_") and normalized[1:] in cls.TYPE_MAPPING:
        # pg catalog array aliases like _int4, _text
        is_array = True
        normalized = normalized[1:]

    mapped: str | None = None
    if normalized in cls.TYPE_MAPPING:
        mapped = cls.TYPE_MAPPING[normalized]
    else:
        for pg_type, graflo_type in cls.TYPE_MAPPING.items():
            if pg_type in normalized or normalized in pg_type:
                logger.debug(
                    f"Mapped PostgreSQL type '{postgres_type}' to graflo type "
                    f"'{graflo_type}' (partial match with '{pg_type}')"
                )
                mapped = graflo_type
                break

    if mapped is None:
        logger.warning(
            f"Unknown PostgreSQL type '{postgres_type}', defaulting to STRING"
        )
        mapped = "STRING"

    if is_array:
        if mapped in {
            "INT",
            "FLOAT",
            "BOOL",
            "STRING",
            "DATETIME",
        }:
            return "LIST", mapped
        # Do not invent a wrong scalar item_type
        logger.debug(
            "PostgreSQL array type '%s' mapped without reliable item_type",
            postgres_type,
        )
        return "LIST", None

    return mapped, None

map_type(postgres_type) classmethod

Map PostgreSQL type to graflo Field type.

Array types (integer[], text[], …) map to LIST; use :meth:map_field when item_type is needed.

Parameters:

Name Type Description Default
postgres_type str

PostgreSQL type name (e.g., 'int4', 'varchar', 'timestamp')

 required

Returns:

Name Type Description
str str

graflo Field type (INT, FLOAT, BOOL, STRING, LIST, …)
Source code in graflo/db/postgres/types.py 
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
@classmethod
def map_type(cls, postgres_type: str) -> str:
    """Map PostgreSQL type to graflo Field type.

    Array types (``integer[]``, ``text[]``, …) map to ``LIST``; use
    :meth:`map_field` when ``item_type`` is needed.

    Args:
        postgres_type: PostgreSQL type name (e.g., 'int4', 'varchar', 'timestamp')

    Returns:
        str: graflo Field type (INT, FLOAT, BOOL, STRING, LIST, …)
    """
    field_type, _item_type = cls.map_field(postgres_type)
    return field_type