Skip to content

graflo.util.onto

Utility ontology classes for resource patterns and configurations.

This module provides data classes for managing resource patterns (files and database tables) and configurations used throughout the system. These classes support resource discovery, pattern matching, and configuration management.

Key Components
  • ResourcePattern: Abstract base class for resource patterns
  • FilePattern: Configuration for file pattern matching
  • TablePattern: Configuration for database table pattern matching
  • Patterns: Collection of named resource patterns with connection management

FilePattern dataclass

Bases: ResourcePattern

Pattern for matching files.

Attributes:

Name Type Description
regex str | None

Regular expression pattern for matching filenames
sub_path None | Path

Path to search for matching files (default: "./")
date_field str | None

Name of the date field to filter on (for date-based filtering)
date_filter str | None

SQL-style date filter condition (e.g., "> '2020-10-10'")
date_range_start str | None

Start date for range filtering (e.g., "2015-11-11")
date_range_days int | None

Number of days after start date (used with date_range_start)
Source code in graflo/util/onto.py 
 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
@dataclasses.dataclass
class FilePattern(ResourcePattern):
    """Pattern for matching files.

    Attributes:
        regex: Regular expression pattern for matching filenames
        sub_path: Path to search for matching files (default: "./")
        date_field: Name of the date field to filter on (for date-based filtering)
        date_filter: SQL-style date filter condition (e.g., "> '2020-10-10'")
        date_range_start: Start date for range filtering (e.g., "2015-11-11")
        date_range_days: Number of days after start date (used with date_range_start)
    """

    class _(BaseDataclass.Meta):
        tag = "file"

    regex: str | None = None
    sub_path: None | pathlib.Path = dataclasses.field(
        default_factory=lambda: pathlib.Path("./")
    )
    date_field: str | None = None
    date_filter: str | None = None
    date_range_start: str | None = None
    date_range_days: int | None = None

    def __post_init__(self):
        """Initialize and validate the file pattern.

        Ensures that sub_path is a Path object and is not None.
        """
        if not isinstance(self.sub_path, pathlib.Path):
            self.sub_path = pathlib.Path(self.sub_path)
        assert self.sub_path is not None
        # Validate date filtering parameters (note: date filtering for files is not yet implemented)
        if (self.date_filter or self.date_range_start) and not self.date_field:
            raise ValueError(
                "date_field is required when using date_filter or date_range_start"
            )
        if self.date_range_days is not None and not self.date_range_start:
            raise ValueError("date_range_start is required when using date_range_days")

    def matches(self, filename: str) -> bool:
        """Check if pattern matches a filename.

        Args:
            filename: Filename to match

        Returns:
            bool: True if pattern matches
        """
        if self.regex is None:
            return False
        return bool(re.match(self.regex, filename))

    def get_resource_type(self) -> ResourceType:
        """Get resource type.

        FilePattern always represents a FILE resource type.
        The specific file format (CSV, JSON, JSONL, Parquet, etc.) is
        automatically detected by the loader based on file extensions.
        """
        return ResourceType.FILE

__post_init__()

Initialize and validate the file pattern.

Ensures that sub_path is a Path object and is not None.

Source code in graflo/util/onto.py 
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
def __post_init__(self):
    """Initialize and validate the file pattern.

    Ensures that sub_path is a Path object and is not None.
    """
    if not isinstance(self.sub_path, pathlib.Path):
        self.sub_path = pathlib.Path(self.sub_path)
    assert self.sub_path is not None
    # Validate date filtering parameters (note: date filtering for files is not yet implemented)
    if (self.date_filter or self.date_range_start) and not self.date_field:
        raise ValueError(
            "date_field is required when using date_filter or date_range_start"
        )
    if self.date_range_days is not None and not self.date_range_start:
        raise ValueError("date_range_start is required when using date_range_days")

get_resource_type()

Get resource type.

FilePattern always represents a FILE resource type. The specific file format (CSV, JSON, JSONL, Parquet, etc.) is automatically detected by the loader based on file extensions.

Source code in graflo/util/onto.py 
137
138
139
140
141
142
143
144
def get_resource_type(self) -> ResourceType:
    """Get resource type.

    FilePattern always represents a FILE resource type.
    The specific file format (CSV, JSON, JSONL, Parquet, etc.) is
    automatically detected by the loader based on file extensions.
    """
    return ResourceType.FILE

matches(filename)

Check if pattern matches a filename.

Parameters:

Name Type Description Default
filename str

Filename to match

 required

Returns:

Name Type Description
bool bool

True if pattern matches
Source code in graflo/util/onto.py 
124
125
126
127
128
129
130
131
132
133
134
135
def matches(self, filename: str) -> bool:
    """Check if pattern matches a filename.

    Args:
        filename: Filename to match

    Returns:
        bool: True if pattern matches
    """
    if self.regex is None:
        return False
    return bool(re.match(self.regex, filename))

Patterns dataclass

Bases: BaseDataclass

Collection of named resource patterns with connection management.

This class manages a collection of resource patterns (files or tables), each associated with a name. It efficiently handles PostgreSQL connections by grouping tables that share the same connection configuration.

The constructor accepts: - resource_mapping: dict mapping resource_name -> (file_path or table_name) - postgres_connections: dict mapping config_key -> PostgresConfig where config_key identifies a connection configuration - postgres_tables: dict mapping table_name -> (config_key, schema_name, table_name)

Attributes:

Name Type Description
patterns dict[str, TablePattern | FilePattern]

Dictionary mapping resource names to ResourcePattern instances
postgres_configs dict[tuple[str, str | None], Any]

Dictionary mapping (config_key, schema_name) to PostgresConfig
postgres_table_configs dict[str, tuple[str, str | None, str]]

Dictionary mapping resource_name to (config_key, schema_name, table_name)
Source code in graflo/util/onto.py 
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
@dataclasses.dataclass
class Patterns(BaseDataclass):
    """Collection of named resource patterns with connection management.

    This class manages a collection of resource patterns (files or tables),
    each associated with a name. It efficiently handles PostgreSQL connections
    by grouping tables that share the same connection configuration.

    The constructor accepts:
    - resource_mapping: dict mapping resource_name -> (file_path or table_name)
    - postgres_connections: dict mapping config_key -> PostgresConfig
      where config_key identifies a connection configuration
    - postgres_tables: dict mapping table_name -> (config_key, schema_name, table_name)

    Attributes:
        patterns: Dictionary mapping resource names to ResourcePattern instances
        postgres_configs: Dictionary mapping (config_key, schema_name) to PostgresConfig
        postgres_table_configs: Dictionary mapping resource_name to (config_key, schema_name, table_name)
    """

    patterns: dict[str, TablePattern | FilePattern] = dataclasses.field(
        default_factory=dict
    )
    postgres_configs: dict[tuple[str, str | None], Any] = dataclasses.field(
        default_factory=dict, metadata={"exclude": True}
    )
    postgres_table_configs: dict[str, tuple[str, str | None, str]] = dataclasses.field(
        default_factory=dict, metadata={"exclude": True}
    )
    # Initialization parameters (not stored as fields, excluded from serialization)
    # Use Any for _postgres_connections to avoid type evaluation issues with dataclass_wizard
    _resource_mapping: dict[str, str | tuple[str, str]] | None = dataclasses.field(
        default=None, repr=False, compare=False, metadata={"exclude": True}
    )
    _postgres_connections: dict[str, Any] | None = dataclasses.field(
        default=None, repr=False, compare=False, metadata={"exclude": True}
    )
    _postgres_tables: dict[str, tuple[str, str | None, str]] | None = dataclasses.field(
        default=None, repr=False, compare=False, metadata={"exclude": True}
    )

    def __post_init__(self):
        """Initialize Patterns from resource mappings and PostgreSQL configurations."""
        # Store PostgreSQL connection configs
        if self._postgres_connections:
            for config_key, config in self._postgres_connections.items():
                if config is not None:
                    schema_name = config.schema_name
                    self.postgres_configs[(config_key, schema_name)] = config

        # Process resource mappings
        if self._resource_mapping:
            for resource_name, resource_spec in self._resource_mapping.items():
                if isinstance(resource_spec, str):
                    # File path - create FilePattern
                    file_path = pathlib.Path(resource_spec)
                    pattern = FilePattern(
                        regex=f"^{re.escape(file_path.name)}$",
                        sub_path=file_path.parent,
                        resource_name=resource_name,
                    )
                    self.patterns[resource_name] = pattern
                elif isinstance(resource_spec, tuple) and len(resource_spec) == 2:
                    # (config_key, table_name) tuple - create TablePattern
                    config_key, table_name = resource_spec
                    # Find the schema_name from the config
                    config = (
                        self._postgres_connections.get(config_key)
                        if self._postgres_connections
                        else None
                    )
                    schema_name = config.schema_name if config else None

                    pattern = TablePattern(
                        table_name=table_name,
                        schema_name=schema_name,
                        resource_name=resource_name,
                    )
                    self.patterns[resource_name] = pattern
                    # Store the config mapping
                    self.postgres_table_configs[resource_name] = (
                        config_key,
                        schema_name,
                        table_name,
                    )

        # Process explicit postgres_tables mapping
        if self._postgres_tables:
            for table_name, (
                config_key,
                schema_name,
                actual_table_name,
            ) in self._postgres_tables.items():
                pattern = TablePattern(
                    table_name=actual_table_name,
                    schema_name=schema_name,
                    resource_name=table_name,
                )
                self.patterns[table_name] = pattern
                self.postgres_table_configs[table_name] = (
                    config_key,
                    schema_name,
                    actual_table_name,
                )

    def add_file_pattern(self, name: str, file_pattern: FilePattern):
        """Add a file pattern to the collection.

        Args:
            name: Name of the pattern
            file_pattern: FilePattern instance
        """
        self.patterns[name] = file_pattern

    def add_table_pattern(self, name: str, table_pattern: TablePattern):
        """Add a table pattern to the collection.

        Args:
            name: Name of the pattern
            table_pattern: TablePattern instance
        """
        self.patterns[name] = table_pattern

    def get_postgres_config(self, resource_name: str) -> Any:
        """Get PostgreSQL connection config for a resource.

        Args:
            resource_name: Name of the resource

        Returns:
            PostgresConfig if resource is a PostgreSQL table, None otherwise
        """
        if resource_name in self.postgres_table_configs:
            config_key, schema_name, _ = self.postgres_table_configs[resource_name]
            return self.postgres_configs.get((config_key, schema_name))
        return None

    def get_resource_type(self, resource_name: str) -> ResourceType | None:
        """Get the resource type for a resource name.

        Args:
            resource_name: Name of the resource

        Returns:
            ResourceType enum value or None if not found
        """
        if resource_name in self.patterns:
            return self.patterns[resource_name].get_resource_type()
        return None

    def get_table_info(self, resource_name: str) -> tuple[str, str | None] | None:
        """Get table name and schema for a PostgreSQL table resource.

        Args:
            resource_name: Name of the resource

        Returns:
            Tuple of (table_name, schema_name) or None if not a table resource
        """
        if resource_name in self.postgres_table_configs:
            _, schema_name, table_name = self.postgres_table_configs[resource_name]
            return (table_name, schema_name)
        return None

__post_init__()

Initialize Patterns from resource mappings and PostgreSQL configurations.

Source code in graflo/util/onto.py 
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
def __post_init__(self):
    """Initialize Patterns from resource mappings and PostgreSQL configurations."""
    # Store PostgreSQL connection configs
    if self._postgres_connections:
        for config_key, config in self._postgres_connections.items():
            if config is not None:
                schema_name = config.schema_name
                self.postgres_configs[(config_key, schema_name)] = config

    # Process resource mappings
    if self._resource_mapping:
        for resource_name, resource_spec in self._resource_mapping.items():
            if isinstance(resource_spec, str):
                # File path - create FilePattern
                file_path = pathlib.Path(resource_spec)
                pattern = FilePattern(
                    regex=f"^{re.escape(file_path.name)}$",
                    sub_path=file_path.parent,
                    resource_name=resource_name,
                )
                self.patterns[resource_name] = pattern
            elif isinstance(resource_spec, tuple) and len(resource_spec) == 2:
                # (config_key, table_name) tuple - create TablePattern
                config_key, table_name = resource_spec
                # Find the schema_name from the config
                config = (
                    self._postgres_connections.get(config_key)
                    if self._postgres_connections
                    else None
                )
                schema_name = config.schema_name if config else None

                pattern = TablePattern(
                    table_name=table_name,
                    schema_name=schema_name,
                    resource_name=resource_name,
                )
                self.patterns[resource_name] = pattern
                # Store the config mapping
                self.postgres_table_configs[resource_name] = (
                    config_key,
                    schema_name,
                    table_name,
                )

    # Process explicit postgres_tables mapping
    if self._postgres_tables:
        for table_name, (
            config_key,
            schema_name,
            actual_table_name,
        ) in self._postgres_tables.items():
            pattern = TablePattern(
                table_name=actual_table_name,
                schema_name=schema_name,
                resource_name=table_name,
            )
            self.patterns[table_name] = pattern
            self.postgres_table_configs[table_name] = (
                config_key,
                schema_name,
                actual_table_name,
            )

add_file_pattern(name, file_pattern)

Add a file pattern to the collection.

Parameters:

Name Type Description Default
name str

Name of the pattern

 required
file_pattern FilePattern

FilePattern instance

 required
Source code in graflo/util/onto.py 
365
366
367
368
369
370
371
372
def add_file_pattern(self, name: str, file_pattern: FilePattern):
    """Add a file pattern to the collection.

    Args:
        name: Name of the pattern
        file_pattern: FilePattern instance
    """
    self.patterns[name] = file_pattern

add_table_pattern(name, table_pattern)

Add a table pattern to the collection.

Parameters:

Name Type Description Default
name str

Name of the pattern

 required
table_pattern TablePattern

TablePattern instance

 required
Source code in graflo/util/onto.py 
374
375
376
377
378
379
380
381
def add_table_pattern(self, name: str, table_pattern: TablePattern):
    """Add a table pattern to the collection.

    Args:
        name: Name of the pattern
        table_pattern: TablePattern instance
    """
    self.patterns[name] = table_pattern

get_postgres_config(resource_name)

Get PostgreSQL connection config for a resource.

Parameters:

Name Type Description Default
resource_name str

Name of the resource

 required

Returns:

Type Description
Any

PostgresConfig if resource is a PostgreSQL table, None otherwise
Source code in graflo/util/onto.py 
383
384
385
386
387
388
389
390
391
392
393
394
395
def get_postgres_config(self, resource_name: str) -> Any:
    """Get PostgreSQL connection config for a resource.

    Args:
        resource_name: Name of the resource

    Returns:
        PostgresConfig if resource is a PostgreSQL table, None otherwise
    """
    if resource_name in self.postgres_table_configs:
        config_key, schema_name, _ = self.postgres_table_configs[resource_name]
        return self.postgres_configs.get((config_key, schema_name))
    return None

get_resource_type(resource_name)

Get the resource type for a resource name.

Parameters:

Name Type Description Default
resource_name str

Name of the resource

 required

Returns:

Type Description
ResourceType | None

ResourceType enum value or None if not found
Source code in graflo/util/onto.py 
397
398
399
400
401
402
403
404
405
406
407
408
def get_resource_type(self, resource_name: str) -> ResourceType | None:
    """Get the resource type for a resource name.

    Args:
        resource_name: Name of the resource

    Returns:
        ResourceType enum value or None if not found
    """
    if resource_name in self.patterns:
        return self.patterns[resource_name].get_resource_type()
    return None

get_table_info(resource_name)

Get table name and schema for a PostgreSQL table resource.

Parameters:

Name Type Description Default
resource_name str

Name of the resource

 required

Returns:

Type Description
tuple[str, str | None] | None

Tuple of (table_name, schema_name) or None if not a table resource
Source code in graflo/util/onto.py 
410
411
412
413
414
415
416
417
418
419
420
421
422
def get_table_info(self, resource_name: str) -> tuple[str, str | None] | None:
    """Get table name and schema for a PostgreSQL table resource.

    Args:
        resource_name: Name of the resource

    Returns:
        Tuple of (table_name, schema_name) or None if not a table resource
    """
    if resource_name in self.postgres_table_configs:
        _, schema_name, table_name = self.postgres_table_configs[resource_name]
        return (table_name, schema_name)
    return None

ResourcePattern dataclass

Bases: BaseDataclass, ABC

Abstract base class for resource patterns (files or tables).

Provides common API for pattern matching and resource identification. All concrete pattern types inherit from this class.

Attributes:

Name Type Description
resource_name str | None

Name of the resource this pattern matches
Source code in graflo/util/onto.py 
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
@dataclasses.dataclass
class ResourcePattern(BaseDataclass, abc.ABC):
    """Abstract base class for resource patterns (files or tables).

    Provides common API for pattern matching and resource identification.
    All concrete pattern types inherit from this class.

    Attributes:
        resource_name: Name of the resource this pattern matches
    """

    resource_name: str | None = None

    @abc.abstractmethod
    def matches(self, resource_identifier: str) -> bool:
        """Check if pattern matches a resource identifier.

        Args:
            resource_identifier: Identifier to match (filename or table name)

        Returns:
            bool: True if pattern matches
        """
        pass

    @abc.abstractmethod
    def get_resource_type(self) -> ResourceType:
        """Get the type of resource this pattern matches.

        Returns:
            ResourceType: Resource type enum value
        """
        pass

get_resource_type() abstractmethod

Get the type of resource this pattern matches.

Returns:

Name Type Description
ResourceType ResourceType

Resource type enum value
Source code in graflo/util/onto.py 
73
74
75
76
77
78
79
80
@abc.abstractmethod
def get_resource_type(self) -> ResourceType:
    """Get the type of resource this pattern matches.

    Returns:
        ResourceType: Resource type enum value
    """
    pass

matches(resource_identifier) abstractmethod

Check if pattern matches a resource identifier.

Parameters:

Name Type Description Default
resource_identifier str

Identifier to match (filename or table name)

 required

Returns:

Name Type Description
bool bool

True if pattern matches
Source code in graflo/util/onto.py 
61
62
63
64
65
66
67
68
69
70
71
@abc.abstractmethod
def matches(self, resource_identifier: str) -> bool:
    """Check if pattern matches a resource identifier.

    Args:
        resource_identifier: Identifier to match (filename or table name)

    Returns:
        bool: True if pattern matches
    """
    pass

ResourceType

Bases: BaseEnum

Resource types for data sources.

Resource types distinguish between different data source categories. File type detection (CSV, JSON, JSONL, Parquet, etc.) is handled automatically by the loader based on file extensions.

Attributes:

Name Type Description
FILE

File-based data source (any format: CSV, JSON, JSONL, Parquet, etc.)
SQL_TABLE

SQL database table (e.g., PostgreSQL table)
Source code in graflo/util/onto.py 
32
33
34
35
36
37
38
39
40
41
42
43
44
45
class ResourceType(BaseEnum):
    """Resource types for data sources.

    Resource types distinguish between different data source categories.
    File type detection (CSV, JSON, JSONL, Parquet, etc.) is handled
    automatically by the loader based on file extensions.

    Attributes:
        FILE: File-based data source (any format: CSV, JSON, JSONL, Parquet, etc.)
        SQL_TABLE: SQL database table (e.g., PostgreSQL table)
    """

    FILE = "file"
    SQL_TABLE = "sql_table"

TablePattern dataclass

Bases: ResourcePattern

Pattern for matching database tables.

Attributes:

Name Type Description
table_name str

Exact table name or regex pattern
schema_name str | None

Schema name (optional, defaults to public)
database str | None

Database name (optional)
date_field str | None

Name of the date field to filter on (for date-based filtering)
date_filter str | None

SQL-style date filter condition (e.g., "> '2020-10-10'")
date_range_start str | None

Start date for range filtering (e.g., "2015-11-11")
date_range_days int | None

Number of days after start date (used with date_range_start)
Source code in graflo/util/onto.py 
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
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
@dataclasses.dataclass
class TablePattern(ResourcePattern):
    """Pattern for matching database tables.

    Attributes:
        table_name: Exact table name or regex pattern
        schema_name: Schema name (optional, defaults to public)
        database: Database name (optional)
        date_field: Name of the date field to filter on (for date-based filtering)
        date_filter: SQL-style date filter condition (e.g., "> '2020-10-10'")
        date_range_start: Start date for range filtering (e.g., "2015-11-11")
        date_range_days: Number of days after start date (used with date_range_start)
    """

    class _(BaseDataclass.Meta):
        tag = "table"

    table_name: str = ""
    schema_name: str | None = None
    database: str | None = None
    date_field: str | None = None
    date_filter: str | None = None
    date_range_start: str | None = None
    date_range_days: int | None = None

    def __post_init__(self):
        """Validate table pattern after initialization."""
        if not self.table_name:
            raise ValueError("table_name is required for TablePattern")
        # Validate date filtering parameters
        if (self.date_filter or self.date_range_start) and not self.date_field:
            raise ValueError(
                "date_field is required when using date_filter or date_range_start"
            )
        if self.date_range_days is not None and not self.date_range_start:
            raise ValueError("date_range_start is required when using date_range_days")

    def matches(self, table_identifier: str) -> bool:
        """Check if pattern matches a table name.

        Args:
            table_identifier: Table name to match (format: schema.table or just table)

        Returns:
            bool: True if pattern matches
        """
        if not self.table_name:
            return False

        # Compile regex pattern
        if self.table_name.startswith("^") or self.table_name.endswith("$"):
            # Already a regex pattern
            pattern = re.compile(self.table_name)
        else:
            # Exact match pattern
            pattern = re.compile(f"^{re.escape(self.table_name)}$")

        # Check if table_identifier matches
        if pattern.match(table_identifier):
            return True

        # If schema_name is specified, also check schema.table format
        if self.schema_name:
            full_name = f"{self.schema_name}.{table_identifier}"
            if pattern.match(full_name):
                return True

        return False

    def get_resource_type(self) -> ResourceType:
        """Get resource type."""
        return ResourceType.SQL_TABLE

    def build_where_clause(self) -> str:
        """Build SQL WHERE clause from date filtering parameters.

        Returns:
            WHERE clause string (without the WHERE keyword) or empty string if no filters
        """
        conditions = []

        if self.date_field:
            if self.date_range_start and self.date_range_days is not None:
                # Range filtering: dt >= start_date AND dt < start_date + interval
                # Example: Ingest for k days after 2015-11-11
                conditions.append(
                    f"\"{self.date_field}\" >= '{self.date_range_start}'::date"
                )
                conditions.append(
                    f"\"{self.date_field}\" < '{self.date_range_start}'::date + INTERVAL '{self.date_range_days} days'"
                )
            elif self.date_filter:
                # Direct filter: dt > 2020-10-10 or dt > '2020-10-10'
                # The date_filter should include the operator and value
                # If value doesn't have quotes, add them
                filter_parts = self.date_filter.strip().split(None, 1)
                if len(filter_parts) == 2:
                    operator, value = filter_parts
                    # Add quotes if not already present and value looks like a date
                    if not (value.startswith("'") and value.endswith("'")):
                        # Check if it's a date-like string (YYYY-MM-DD format)
                        if len(value) == 10 and value.count("-") == 2:
                            value = f"'{value}'"
                    conditions.append(f'"{self.date_field}" {operator} {value}')
                else:
                    # If format is unexpected, use as-is
                    conditions.append(f'"{self.date_field}" {self.date_filter}')

        if conditions:
            return " AND ".join(conditions)
        return ""

__post_init__()

Validate table pattern after initialization.

Source code in graflo/util/onto.py 
172
173
174
175
176
177
178
179
180
181
182
def __post_init__(self):
    """Validate table pattern after initialization."""
    if not self.table_name:
        raise ValueError("table_name is required for TablePattern")
    # Validate date filtering parameters
    if (self.date_filter or self.date_range_start) and not self.date_field:
        raise ValueError(
            "date_field is required when using date_filter or date_range_start"
        )
    if self.date_range_days is not None and not self.date_range_start:
        raise ValueError("date_range_start is required when using date_range_days")

build_where_clause()

Build SQL WHERE clause from date filtering parameters.

Returns:

Type Description
str

WHERE clause string (without the WHERE keyword) or empty string if no filters
Source code in graflo/util/onto.py 
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
def build_where_clause(self) -> str:
    """Build SQL WHERE clause from date filtering parameters.

    Returns:
        WHERE clause string (without the WHERE keyword) or empty string if no filters
    """
    conditions = []

    if self.date_field:
        if self.date_range_start and self.date_range_days is not None:
            # Range filtering: dt >= start_date AND dt < start_date + interval
            # Example: Ingest for k days after 2015-11-11
            conditions.append(
                f"\"{self.date_field}\" >= '{self.date_range_start}'::date"
            )
            conditions.append(
                f"\"{self.date_field}\" < '{self.date_range_start}'::date + INTERVAL '{self.date_range_days} days'"
            )
        elif self.date_filter:
            # Direct filter: dt > 2020-10-10 or dt > '2020-10-10'
            # The date_filter should include the operator and value
            # If value doesn't have quotes, add them
            filter_parts = self.date_filter.strip().split(None, 1)
            if len(filter_parts) == 2:
                operator, value = filter_parts
                # Add quotes if not already present and value looks like a date
                if not (value.startswith("'") and value.endswith("'")):
                    # Check if it's a date-like string (YYYY-MM-DD format)
                    if len(value) == 10 and value.count("-") == 2:
                        value = f"'{value}'"
                conditions.append(f'"{self.date_field}" {operator} {value}')
            else:
                # If format is unexpected, use as-is
                conditions.append(f'"{self.date_field}" {self.date_filter}')

    if conditions:
        return " AND ".join(conditions)
    return ""

get_resource_type()

Get resource type.

Source code in graflo/util/onto.py 
216
217
218
def get_resource_type(self) -> ResourceType:
    """Get resource type."""
    return ResourceType.SQL_TABLE

matches(table_identifier)

Check if pattern matches a table name.

Parameters:

Name Type Description Default
table_identifier str

Table name to match (format: schema.table or just table)

 required

Returns:

Name Type Description
bool bool

True if pattern matches
Source code in graflo/util/onto.py 
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
def matches(self, table_identifier: str) -> bool:
    """Check if pattern matches a table name.

    Args:
        table_identifier: Table name to match (format: schema.table or just table)

    Returns:
        bool: True if pattern matches
    """
    if not self.table_name:
        return False

    # Compile regex pattern
    if self.table_name.startswith("^") or self.table_name.endswith("$"):
        # Already a regex pattern
        pattern = re.compile(self.table_name)
    else:
        # Exact match pattern
        pattern = re.compile(f"^{re.escape(self.table_name)}$")

    # Check if table_identifier matches
    if pattern.match(table_identifier):
        return True

    # If schema_name is specified, also check schema.table format
    if self.schema_name:
        full_name = f"{self.schema_name}.{table_identifier}"
        if pattern.match(full_name):
            return True

    return False