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
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.

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

        Returns:
            str: graflo Field type (INT, FLOAT, BOOL, or STRING)
        """
        # Normalize type name: lowercase and remove length specifications
        normalized = postgres_type.lower().strip()

        # Remove length specifications like (255) or (10,2)
        if "(" in normalized:
            normalized = normalized.split("(")[0].strip()

        # Check direct mapping
        if normalized in cls.TYPE_MAPPING:
            return cls.TYPE_MAPPING[normalized]

        # Check for partial matches (e.g., "character varying" might be stored as "varying")
        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 '{graflo_type}' "
                    f"(partial match with '{pg_type}')"
                )
                return graflo_type

        # Default to STRING for unknown types
        logger.warning(
            f"Unknown PostgreSQL type '{postgres_type}', defaulting to STRING"
        )
        return "STRING"

    @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 
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
@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 
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
@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_type(postgres_type) classmethod

Map PostgreSQL type to graflo Field type.

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, or STRING)
Source code in graflo/db/postgres/types.py 
 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
@classmethod
def map_type(cls, postgres_type: str) -> str:
    """Map PostgreSQL type to graflo Field type.

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

    Returns:
        str: graflo Field type (INT, FLOAT, BOOL, or STRING)
    """
    # Normalize type name: lowercase and remove length specifications
    normalized = postgres_type.lower().strip()

    # Remove length specifications like (255) or (10,2)
    if "(" in normalized:
        normalized = normalized.split("(")[0].strip()

    # Check direct mapping
    if normalized in cls.TYPE_MAPPING:
        return cls.TYPE_MAPPING[normalized]

    # Check for partial matches (e.g., "character varying" might be stored as "varying")
    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 '{graflo_type}' "
                f"(partial match with '{pg_type}')"
            )
            return graflo_type

    # Default to STRING for unknown types
    logger.warning(
        f"Unknown PostgreSQL type '{postgres_type}', defaulting to STRING"
    )
    return "STRING"