mirror of
https://github.com/semantica-agi/semantica.git
synced 2026-08-29 04:26:20 +00:00
* Added Parquet ingest support (#234) * docs: Add Parquet ingestion support to CHANGELOG - Add comprehensive changelog entry for PR #548 - Document ParquetIngestor class and key features - Include author credit (@Luffy2208) and PR reference - Follow existing changelog format and structure --------- Co-authored-by: KaifAhmad1 <kaifahmad087@gmail.com>
This commit is contained in:
Luffy2208
@@ -10,6 +10,17 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
|
||||
## [Unreleased]
|
||||
|
||||
### Added
|
||||
- **Parquet File Ingestion Support** (#548) by @Luffy2208
|
||||
- Added ParquetIngestor class with PyArrow backend
|
||||
- Single file and partitioned directory ingestion
|
||||
- Schema and metadata extraction capabilities
|
||||
- Selective column reading with memory efficiency
|
||||
- Hive-style partition discovery support
|
||||
- Unified dispatch integration
|
||||
- Optional dependency management (ingest-parquet extra)
|
||||
- Comprehensive test coverage (32/32 tests passing)
|
||||
|
||||
**Ontology Hub** (part of #517)
|
||||
|
||||
**Ontology Hub** (part of #517)
|
||||
|
||||
|
||||
@@ -12,6 +12,7 @@ The **Ingest Module** is the entry point for loading data into Semantica. It pro
|
||||
|
||||
**Data ingestion** is the process of loading data from various sources into Semantica for processing. The ingest module handles:
|
||||
- **File Systems**: Local files, cloud storage (S3, GCS, Azure)
|
||||
- **Analytics Files**: Apache Parquet files and partitioned datasets
|
||||
- **Web Content**: Websites, RSS feeds, APIs
|
||||
- **Streams**: Real-time data from Kafka, RabbitMQ, etc.
|
||||
- **Databases**: SQL, NoSQL, and cloud data warehouses including Snowflake
|
||||
@@ -74,6 +75,12 @@ The **Ingest Module** is the entry point for loading data into Semantica. It pro
|
||||
|
||||
Ingest tables and query results from SQL, NoSQL, and cloud data warehouses including Snowflake
|
||||
|
||||
- :material-table:{ .lg .middle } **Parquet Datasets**
|
||||
|
||||
---
|
||||
|
||||
Read Parquet files, schemas, metadata, and Hive-style partitioned directories
|
||||
|
||||
</div>
|
||||
|
||||
!!! tip "When to Use"
|
||||
@@ -118,6 +125,19 @@ Handles file systems and object storage.
|
||||
| `ingest_file(path)` | Process single file |
|
||||
| `ingest_directory(path)` | Process folder |
|
||||
|
||||
### ParquetIngestor
|
||||
|
||||
Handles Apache Parquet files and partitioned datasets.
|
||||
|
||||
**Methods:**
|
||||
|
||||
| Method | Description |
|
||||
|--------|-------------|
|
||||
| `ingest_file(path, columns=None, limit=None)` | Read a Parquet file |
|
||||
| `ingest_directory(path, columns=None, limit=None)` | Read a partitioned Parquet directory |
|
||||
| `extract_schema(path)` | Extract column names, types, nullability, and schema metadata |
|
||||
| `extract_metadata(path)` | Extract row counts, row groups, compression, and partition info |
|
||||
|
||||
### WebIngestor
|
||||
|
||||
Handles web content.
|
||||
@@ -213,10 +233,33 @@ from semantica.ingest import ingest
|
||||
|
||||
# Auto-detect source type
|
||||
ingest("doc.pdf", source_type="file")
|
||||
ingest("events.parquet") # Auto-detects Parquet
|
||||
ingest("https://google.com", source_type="web")
|
||||
ingest("kafka://topic", source_type="stream")
|
||||
```
|
||||
|
||||
### Parquet Dataset Ingestion
|
||||
|
||||
```python
|
||||
from semantica.ingest import ParquetIngestor, ingest_parquet
|
||||
|
||||
ingestor = ParquetIngestor()
|
||||
|
||||
# Read selected columns from a local Parquet file
|
||||
events = ingestor.ingest_file(
|
||||
"events.parquet",
|
||||
columns=["event_id", "event_type"],
|
||||
limit=1000,
|
||||
)
|
||||
|
||||
# Inspect schema and metadata without reading rows
|
||||
schema = ingestor.extract_schema("events.parquet")
|
||||
metadata = ingestor.extract_metadata("events.parquet")
|
||||
|
||||
# Read a Hive-style partitioned directory such as country=US/year=2026/
|
||||
partitioned = ingest_parquet("./warehouse/events", method="directory")
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
## Configuration
|
||||
@@ -236,7 +279,7 @@ ingest:
|
||||
web:
|
||||
user_agent: "MyBot"
|
||||
rate_limit: 1.0 # seconds
|
||||
|
||||
|
||||
files:
|
||||
max_size: 100MB
|
||||
allowed_extensions: [.pdf, .txt, .md]
|
||||
@@ -289,12 +332,12 @@ data = ingestor.ingest_snowflake_table("CUSTOMERS")
|
||||
|
||||
# 3. Or run custom query
|
||||
results = ingestor.execute_snowflake_query("""
|
||||
SELECT
|
||||
CUSTOMER_ID,
|
||||
NAME,
|
||||
EMAIL,
|
||||
CREATED_AT
|
||||
FROM CUSTOMERS
|
||||
SELECT
|
||||
CUSTOMER_ID,
|
||||
NAME,
|
||||
EMAIL,
|
||||
CREATED_AT
|
||||
FROM CUSTOMERS
|
||||
WHERE CREATED_AT > '2024-01-01'
|
||||
""")
|
||||
|
||||
|
||||
+3
-2
@@ -104,6 +104,7 @@ parse-docling = ["docling>=1.0.0"]
|
||||
# ---- Database Connectors ----
|
||||
db-snowflake = ["snowflake-connector-python>=3.0.0", "cryptography>=3.4.0"]
|
||||
db-arrow = ["pyarrow>=10.0.0"]
|
||||
ingest-parquet = ["pyarrow>=10.0.0"]
|
||||
|
||||
db-all = [
|
||||
"semantica[db-snowflake,db-arrow]"
|
||||
@@ -213,8 +214,8 @@ explorer-lite = [
|
||||
|
||||
# Everything (cross-platform — gpu excluded; install semantica[gpu] separately on Linux)
|
||||
all = [
|
||||
"semantica[dev,viz,infra,cloud,monitoring,llm-all,models-huggingface,split-all,graph-all,vectorstore-all,parse-docling,explorer]",
|
||||
"semantica[dev,viz,infra,cloud,monitoring,llm-all,models-huggingface,split-all,graph-all,vectorstore-all,parse-docling,agno]"
|
||||
"semantica[dev,viz,infra,cloud,monitoring,llm-all,models-huggingface,split-all,graph-all,vectorstore-all,parse-docling,ingest-parquet,explorer]",
|
||||
"semantica[dev,viz,infra,cloud,monitoring,llm-all,models-huggingface,split-all,graph-all,vectorstore-all,parse-docling,ingest-parquet,agno]"
|
||||
]
|
||||
|
||||
# ---------------- ENTRYPOINTS ----------------
|
||||
|
||||
@@ -7,12 +7,15 @@ including files, web content, feeds, streams, repositories, emails, and database
|
||||
Algorithms Used:
|
||||
|
||||
File Ingestion:
|
||||
- File Type Detection: Multi-method detection (extension-based, MIME type, magic number analysis)
|
||||
- File Type Detection: Multi-method detection using extension,
|
||||
MIME type, and magic number analysis
|
||||
- Directory Scanning: Recursive directory traversal with filtering
|
||||
- Cloud Storage Integration: AWS S3, Google Cloud Storage, Azure Blob Storage API integration
|
||||
- Cloud Storage Integration: AWS S3, Google Cloud Storage, Azure Blob
|
||||
Storage API integration
|
||||
- File Validation: Size limits, format validation, content verification
|
||||
- Batch Processing: Parallel file processing with progress tracking
|
||||
- Magic Number Analysis: Binary file signature detection for accurate type identification
|
||||
- Magic Number Analysis: Binary file signature detection for accurate
|
||||
type identification
|
||||
|
||||
Web Ingestion:
|
||||
- HTTP Request Handling: GET/POST requests with retry logic and error handling
|
||||
@@ -46,9 +49,11 @@ Repository Ingestion:
|
||||
- Git Operations: Repository cloning, branch checking, commit traversal
|
||||
- Code Extraction: File content extraction with language detection
|
||||
- Commit Analysis: Git log parsing, diff analysis, statistics calculation
|
||||
- Language Detection: File extension and content-based programming language identification
|
||||
- Language Detection: File extension and content-based programming
|
||||
language identification
|
||||
- Code Structure Analysis: AST parsing for classes, functions, imports extraction
|
||||
- Dependency Analysis: Package manager file parsing (requirements.txt, package.json, etc.)
|
||||
- Dependency Analysis: Package manager file parsing
|
||||
(requirements.txt, package.json, etc.)
|
||||
- Documentation Extraction: README, docstring, and comment extraction
|
||||
|
||||
Email Ingestion:
|
||||
@@ -61,7 +66,8 @@ Email Ingestion:
|
||||
- Link Extraction: URL extraction from email HTML content
|
||||
|
||||
Database Ingestion:
|
||||
- Database Connection: SQLAlchemy-based connection management with connection pooling
|
||||
- Database Connection: SQLAlchemy-based connection management with
|
||||
connection pooling
|
||||
- SQL Query Execution: Parameterized query execution with result set processing
|
||||
- Schema Introspection: Database schema analysis and table/column discovery
|
||||
- Data Type Conversion: Database-specific type to Python type conversion
|
||||
@@ -87,6 +93,7 @@ Main Classes:
|
||||
- EmailIngestor: Email protocol handling
|
||||
- DBIngestor: Database export handling
|
||||
- OntologyIngestor: Ontology file processing
|
||||
- ParquetIngestor: Apache Parquet file and partitioned dataset processing
|
||||
- MethodRegistry: Registry for custom ingestion methods
|
||||
- IngestConfig: Configuration manager for ingest module
|
||||
|
||||
@@ -100,6 +107,7 @@ Convenience Functions:
|
||||
- ingest_email: Email ingestion wrapper
|
||||
- ingest_database: Database ingestion wrapper
|
||||
- ingest_ontology: Ontology ingestion wrapper
|
||||
- ingest_parquet: Parquet ingestion wrapper
|
||||
|
||||
|
||||
Example Usage:
|
||||
@@ -133,6 +141,7 @@ from .methods import (
|
||||
ingest_file,
|
||||
ingest_mcp,
|
||||
ingest_ontology,
|
||||
ingest_parquet,
|
||||
ingest_repository,
|
||||
ingest_stream,
|
||||
ingest_web,
|
||||
@@ -192,6 +201,9 @@ _LAZY_EXPORTS: Dict[str, Tuple[str, str]] = {
|
||||
"SnowflakeIngestor": (".snowflake_ingestor", "SnowflakeIngestor"),
|
||||
"SnowflakeData": (".snowflake_ingestor", "SnowflakeData"),
|
||||
"SnowflakeConnector": (".snowflake_ingestor", "SnowflakeConnector"),
|
||||
# Parquet ingestion
|
||||
"ParquetIngestor": (".parquet_ingestor", "ParquetIngestor"),
|
||||
"ParquetData": (".parquet_ingestor", "ParquetData"),
|
||||
}
|
||||
|
||||
_OPTIONAL_DEPENDENCY_MESSAGES = {
|
||||
@@ -211,6 +223,10 @@ _OPTIONAL_DEPENDENCY_MESSAGES = {
|
||||
"Email ingestion requires optional dependency 'beautifulsoup4'. "
|
||||
"Install it before importing EmailIngestor or using ingest_email()."
|
||||
),
|
||||
".parquet_ingestor": (
|
||||
"Parquet ingestion requires optional dependency 'pyarrow'. "
|
||||
"Install it before importing ParquetIngestor or using ingest_parquet()."
|
||||
),
|
||||
}
|
||||
|
||||
|
||||
@@ -225,7 +241,7 @@ def __getattr__(name: str) -> Any:
|
||||
except ModuleNotFoundError as exc:
|
||||
message = _OPTIONAL_DEPENDENCY_MESSAGES.get(module_name)
|
||||
missing_name = getattr(exc, "name", None)
|
||||
if message and missing_name in {"git", "bs4"}:
|
||||
if message and missing_name in {"git", "bs4", "pyarrow"}:
|
||||
raise ImportError(message) from exc
|
||||
raise
|
||||
|
||||
@@ -233,6 +249,7 @@ def __getattr__(name: str) -> Any:
|
||||
globals()[name] = value
|
||||
return value
|
||||
|
||||
|
||||
__all__ = [
|
||||
# File ingestion
|
||||
"FileIngestor",
|
||||
@@ -290,6 +307,9 @@ __all__ = [
|
||||
"SnowflakeIngestor",
|
||||
"SnowflakeData",
|
||||
"SnowflakeConnector",
|
||||
# Parquet ingestion
|
||||
"ParquetIngestor",
|
||||
"ParquetData",
|
||||
# Registry and Methods
|
||||
"MethodRegistry",
|
||||
"method_registry",
|
||||
@@ -302,6 +322,7 @@ __all__ = [
|
||||
"ingest_email",
|
||||
"ingest_database",
|
||||
"ingest_ontology",
|
||||
"ingest_parquet",
|
||||
"ingest_mcp",
|
||||
"get_ingest_method",
|
||||
"list_available_methods",
|
||||
@@ -309,4 +330,3 @@ __all__ = [
|
||||
"IngestConfig",
|
||||
"ingest_config",
|
||||
]
|
||||
|
||||
|
||||
@@ -23,7 +23,6 @@ License: MIT
|
||||
"""
|
||||
|
||||
import mimetypes
|
||||
import os
|
||||
from dataclasses import dataclass, field
|
||||
from datetime import datetime
|
||||
from pathlib import Path
|
||||
@@ -112,7 +111,8 @@ class FileTypeDetector:
|
||||
mimetypes.init()
|
||||
|
||||
self.logger.debug(
|
||||
f"File type detector initialized with {len(self.supported_formats)} supported formats"
|
||||
"File type detector initialized with "
|
||||
f"{len(self.supported_formats)} supported formats"
|
||||
)
|
||||
|
||||
def detect_type(
|
||||
@@ -190,11 +190,12 @@ class FileTypeDetector:
|
||||
magic_numbers = {
|
||||
b"\x25\x50\x44\x46": "pdf", # PDF (binary)
|
||||
b"%PDF": "pdf", # PDF (text header)
|
||||
b"\x50\x4B\x03\x04": "zip", # ZIP, DOCX, XLSX, PPTX (Office Open XML)
|
||||
b"\x89\x50\x4E\x47": "png", # PNG image
|
||||
b"\xFF\xD8\xFF": "jpg", # JPEG image
|
||||
b"\x50\x4b\x03\x04": "zip", # ZIP, DOCX, XLSX, PPTX (Office Open XML)
|
||||
b"\x89\x50\x4e\x47": "png", # PNG image
|
||||
b"\xff\xd8\xff": "jpg", # JPEG image
|
||||
b"\x47\x49\x46\x38": "gif", # GIF image
|
||||
b"PK\x03\x04": "zip", # ZIP (alternative)
|
||||
b"PAR1": "parquet", # Apache Parquet
|
||||
}
|
||||
|
||||
# Check if content starts with any known magic number
|
||||
@@ -516,7 +517,10 @@ class FileIngestor:
|
||||
tracking_id,
|
||||
processed=idx,
|
||||
total=total_files,
|
||||
message=f"Processing file {idx}/{total_files}: {Path(file_info['path']).name}"
|
||||
message=(
|
||||
f"Processing file {idx}/{total_files}: "
|
||||
f"{Path(file_info['path']).name}"
|
||||
),
|
||||
)
|
||||
|
||||
# Track progress via callback if provided
|
||||
|
||||
@@ -6,18 +6,19 @@ This guide demonstrates how to use the ingest module for ingesting data from var
|
||||
|
||||
1. [Basic Usage](#basic-usage)
|
||||
2. [File Ingestion](#file-ingestion)
|
||||
3. [Web Ingestion](#web-ingestion)
|
||||
4. [Feed Ingestion](#feed-ingestion)
|
||||
5. [Stream Ingestion](#stream-ingestion)
|
||||
6. [Repository Ingestion](#repository-ingestion)
|
||||
7. [Email Ingestion](#email-ingestion)
|
||||
8. [Database Ingestion](#database-ingestion)
|
||||
9. [MCP Server Ingestion](#mcp-server-ingestion)
|
||||
10. [Unified Ingestion](#unified-ingestion)
|
||||
11. [Using Methods](#using-methods)
|
||||
12. [Using Registry](#using-registry)
|
||||
13. [Configuration](#configuration)
|
||||
14. [Advanced Examples](#advanced-examples)
|
||||
3. [Parquet Ingestion](#parquet-ingestion)
|
||||
4. [Web Ingestion](#web-ingestion)
|
||||
5. [Feed Ingestion](#feed-ingestion)
|
||||
6. [Stream Ingestion](#stream-ingestion)
|
||||
7. [Repository Ingestion](#repository-ingestion)
|
||||
8. [Email Ingestion](#email-ingestion)
|
||||
9. [Database Ingestion](#database-ingestion)
|
||||
10. [MCP Server Ingestion](#mcp-server-ingestion)
|
||||
11. [Unified Ingestion](#unified-ingestion)
|
||||
12. [Using Methods](#using-methods)
|
||||
13. [Using Registry](#using-registry)
|
||||
14. [Configuration](#configuration)
|
||||
15. [Advanced Examples](#advanced-examples)
|
||||
|
||||
## Basic Usage
|
||||
|
||||
@@ -29,6 +30,9 @@ from semantica.ingest import ingest
|
||||
# Ingest a file (auto-detects source type)
|
||||
result = ingest("document.pdf", source_type="file")
|
||||
|
||||
# Ingest a Parquet file
|
||||
result = ingest("events.parquet")
|
||||
|
||||
# Ingest from web URL
|
||||
result = ingest("https://example.com", source_type="web")
|
||||
|
||||
@@ -142,6 +146,66 @@ with open("document.pdf", "rb") as f:
|
||||
file_type = detector.detect_type("document.pdf", content=content)
|
||||
```
|
||||
|
||||
## Parquet Ingestion
|
||||
|
||||
Parquet ingestion requires PyArrow:
|
||||
|
||||
```bash
|
||||
pip install pyarrow
|
||||
```
|
||||
|
||||
### Single Parquet File
|
||||
|
||||
```python
|
||||
from semantica.ingest import ParquetIngestor, ingest_parquet
|
||||
|
||||
# Using convenience function
|
||||
data = ingest_parquet(
|
||||
"events.parquet",
|
||||
columns=["event_id", "event_type"],
|
||||
limit=1000,
|
||||
)
|
||||
|
||||
# Using class directly
|
||||
ingestor = ParquetIngestor()
|
||||
data = ingestor.ingest_file("events.parquet")
|
||||
|
||||
print(f"Rows returned: {data.row_count}")
|
||||
print(f"Columns: {data.columns}")
|
||||
print(f"Total rows in file: {data.metadata['total_rows']}")
|
||||
```
|
||||
|
||||
### Schema and Metadata Extraction
|
||||
|
||||
```python
|
||||
from semantica.ingest import ParquetIngestor
|
||||
|
||||
ingestor = ParquetIngestor()
|
||||
|
||||
schema = ingestor.extract_schema("events.parquet")
|
||||
metadata = ingestor.extract_metadata("events.parquet")
|
||||
|
||||
print(schema["columns"])
|
||||
print(metadata["compression_codecs"])
|
||||
print(metadata["row_groups"])
|
||||
```
|
||||
|
||||
### Partitioned Parquet Directories
|
||||
|
||||
```python
|
||||
from semantica.ingest import ingest_parquet
|
||||
|
||||
# Reads Hive-style directories such as country=US/year=2026/part-0.parquet
|
||||
data = ingest_parquet(
|
||||
"./warehouse/events",
|
||||
method="directory",
|
||||
columns=["event_id", "event_type", "country", "year"],
|
||||
)
|
||||
|
||||
print(data.metadata["partition_columns"])
|
||||
print(data.metadata["partition_values"])
|
||||
```
|
||||
|
||||
## Web Ingestion
|
||||
|
||||
### Single URL Ingestion
|
||||
@@ -936,6 +1000,7 @@ from semantica.ingest import ingest
|
||||
|
||||
# Auto-detect source type from source
|
||||
result = ingest("document.pdf") # Auto-detects file
|
||||
result = ingest("events.parquet") # Auto-detects Parquet
|
||||
result = ingest("https://example.com") # Auto-detects web
|
||||
result = ingest("https://example.com/feed.xml") # Auto-detects feed
|
||||
result = ingest("postgresql://user:pass@localhost/db") # Auto-detects database
|
||||
@@ -982,7 +1047,8 @@ from semantica.ingest.methods import (
|
||||
ingest_repository,
|
||||
ingest_email,
|
||||
ingest_database,
|
||||
ingest_mcp
|
||||
ingest_mcp,
|
||||
ingest_parquet
|
||||
)
|
||||
|
||||
# File ingestion
|
||||
@@ -1006,6 +1072,9 @@ emails = ingest_email({"host": "imap.example.com", "username": "user", "password
|
||||
# Database ingestion
|
||||
data = ingest_database("postgresql://user:pass@localhost/db", table="users")
|
||||
|
||||
# Parquet ingestion
|
||||
events = ingest_parquet("events.parquet", columns=["event_id"], limit=1000)
|
||||
|
||||
# MCP server ingestion via URL
|
||||
data = ingest_mcp("http://localhost:8000/mcp", method="resources")
|
||||
```
|
||||
@@ -1189,16 +1258,16 @@ from semantica.ingest.methods import ingest_file
|
||||
def custom_pdf_ingestion(source, **kwargs):
|
||||
"""Custom PDF ingestion with special processing."""
|
||||
from semantica.ingest import FileIngestor
|
||||
|
||||
|
||||
ingestor = FileIngestor()
|
||||
file_obj = ingestor.ingest_file(source, **kwargs)
|
||||
|
||||
|
||||
# Custom processing
|
||||
if file_obj.file_type == "pdf":
|
||||
# Add custom metadata
|
||||
file_obj.metadata["processed"] = True
|
||||
file_obj.metadata["custom_field"] = "custom_value"
|
||||
|
||||
|
||||
return file_obj
|
||||
|
||||
# Register custom method
|
||||
@@ -1286,7 +1355,7 @@ for source_type, source_list in sources.items():
|
||||
1. **Parallel Processing**: Use parallel processing for multiple sources
|
||||
```python
|
||||
from concurrent.futures import ThreadPoolExecutor
|
||||
|
||||
|
||||
with ThreadPoolExecutor() as executor:
|
||||
executor.submit(ingest_file, "./documents1")
|
||||
executor.submit(ingest_file, "./documents2")
|
||||
@@ -1329,4 +1398,3 @@ for source_type, source_list in sources.items():
|
||||
for batch in process_in_batches(large_dataset, batch_size=1000):
|
||||
result = ingest(batch)
|
||||
```
|
||||
|
||||
|
||||
+176
-28
@@ -13,6 +13,12 @@ File Ingestion:
|
||||
- "directory": Directory ingestion with recursive scanning
|
||||
- "cloud": Cloud storage ingestion (S3, GCS, Azure)
|
||||
|
||||
Parquet Ingestion:
|
||||
- "file": Single Parquet file ingestion
|
||||
- "directory": Partitioned Parquet directory ingestion
|
||||
- "schema": Parquet schema extraction
|
||||
- "metadata": Parquet file or directory metadata extraction
|
||||
|
||||
Web Ingestion:
|
||||
- "url": Single URL ingestion
|
||||
- "sitemap": Sitemap-based crawling
|
||||
@@ -48,12 +54,15 @@ Database Ingestion:
|
||||
Algorithms Used:
|
||||
|
||||
File Ingestion:
|
||||
- File Type Detection: Multi-method detection (extension-based, MIME type, magic number analysis)
|
||||
- File Type Detection: Multi-method detection using extension,
|
||||
MIME type, and magic number analysis
|
||||
- Directory Scanning: Recursive directory traversal with filtering
|
||||
- Cloud Storage Integration: AWS S3, Google Cloud Storage, Azure Blob Storage API integration
|
||||
- Cloud Storage Integration: AWS S3, Google Cloud Storage, Azure Blob
|
||||
Storage API integration
|
||||
- File Validation: Size limits, format validation, content verification
|
||||
- Batch Processing: Parallel file processing with progress tracking
|
||||
- Magic Number Analysis: Binary file signature detection for accurate type identification
|
||||
- Magic Number Analysis: Binary file signature detection for accurate
|
||||
type identification
|
||||
|
||||
Web Ingestion:
|
||||
- HTTP Request Handling: GET/POST requests with retry logic and error handling
|
||||
@@ -87,9 +96,11 @@ Repository Ingestion:
|
||||
- Git Operations: Repository cloning, branch checking, commit traversal
|
||||
- Code Extraction: File content extraction with language detection
|
||||
- Commit Analysis: Git log parsing, diff analysis, statistics calculation
|
||||
- Language Detection: File extension and content-based programming language identification
|
||||
- Language Detection: File extension and content-based programming
|
||||
language identification
|
||||
- Code Structure Analysis: AST parsing for classes, functions, imports extraction
|
||||
- Dependency Analysis: Package manager file parsing (requirements.txt, package.json, etc.)
|
||||
- Dependency Analysis: Package manager file parsing
|
||||
(requirements.txt, package.json, etc.)
|
||||
- Documentation Extraction: README, docstring, and comment extraction
|
||||
|
||||
Email Ingestion:
|
||||
@@ -102,7 +113,8 @@ Email Ingestion:
|
||||
- Link Extraction: URL extraction from email HTML content
|
||||
|
||||
Database Ingestion:
|
||||
- Database Connection: SQLAlchemy-based connection management with connection pooling
|
||||
- Database Connection: SQLAlchemy-based connection management with
|
||||
connection pooling
|
||||
- SQL Query Execution: Parameterized query execution with result set processing
|
||||
- Schema Introspection: Database schema analysis and table/column discovery
|
||||
- Data Type Conversion: Database-specific type to Python type conversion
|
||||
@@ -125,6 +137,7 @@ Main Functions:
|
||||
- ingest_repository: Repository ingestion wrapper
|
||||
- ingest_email: Email ingestion wrapper
|
||||
- ingest_database: Database ingestion wrapper
|
||||
- ingest_parquet: Parquet ingestion wrapper
|
||||
- ingest: Unified ingestion function with source type dispatch
|
||||
- get_ingest_method: Get ingestion method by name
|
||||
- list_available_methods: List registered methods
|
||||
@@ -142,7 +155,7 @@ Example Usage:
|
||||
from __future__ import annotations
|
||||
|
||||
from pathlib import Path
|
||||
from typing import Any, Callable, Dict, List, Optional, Union
|
||||
from typing import TYPE_CHECKING, Any, Callable, Dict, List, Optional, Union
|
||||
|
||||
from ..utils.exceptions import ConfigurationError, ProcessingError
|
||||
from ..utils.logging import get_logger
|
||||
@@ -150,6 +163,16 @@ from .config import ingest_config
|
||||
from .file_ingestor import FileIngestor, FileObject
|
||||
from .registry import method_registry
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from .db_ingestor import TableData
|
||||
from .email_ingestor import EmailData
|
||||
from .feed_ingestor import FeedData
|
||||
from .mcp_ingestor import MCPData
|
||||
from .ontology_ingestor import OntologyData
|
||||
from .parquet_ingestor import ParquetData
|
||||
from .stream_ingestor import StreamProcessor
|
||||
from .web_ingestor import WebContent
|
||||
|
||||
logger = get_logger("ingest_methods")
|
||||
|
||||
|
||||
@@ -230,6 +253,86 @@ def ingest_file(
|
||||
raise
|
||||
|
||||
|
||||
def ingest_parquet(
|
||||
source: Union[str, Path, List[Union[str, Path]]],
|
||||
method: str = "file",
|
||||
**kwargs,
|
||||
) -> Union[ParquetData, List[ParquetData], Dict[str, Any]]:
|
||||
"""
|
||||
Ingest Apache Parquet files or partitioned directories.
|
||||
|
||||
Args:
|
||||
source: Parquet file path, directory path, or list of paths
|
||||
method: Ingestion method:
|
||||
- "file": Single Parquet file ingestion
|
||||
- "directory": Parquet directory or partitioned dataset ingestion
|
||||
- "schema": Extract schema without reading data
|
||||
- "metadata": Extract file/directory metadata without reading data
|
||||
**kwargs: Additional options passed to ParquetIngestor
|
||||
|
||||
Returns:
|
||||
ParquetData, list of ParquetData, or metadata/schema dictionary
|
||||
|
||||
Examples:
|
||||
>>> from semantica.ingest.methods import ingest_parquet
|
||||
>>> data = ingest_parquet("events.parquet", columns=["id"], limit=100)
|
||||
>>> schema = ingest_parquet("events.parquet", method="schema")
|
||||
>>> dataset = ingest_parquet("./events_by_date", method="directory")
|
||||
"""
|
||||
custom_method = method_registry.get("parquet", method)
|
||||
if custom_method and custom_method != ingest_parquet:
|
||||
try:
|
||||
return custom_method(source, **kwargs)
|
||||
except Exception as e:
|
||||
logger.warning(
|
||||
f"Custom method {method} failed: {e}, falling back to default"
|
||||
)
|
||||
|
||||
try:
|
||||
try:
|
||||
from .parquet_ingestor import ParquetIngestor
|
||||
except ModuleNotFoundError as exc:
|
||||
if _is_missing_dependency(exc, "pyarrow"):
|
||||
raise _missing_optional_dependency(
|
||||
"Parquet ingestion",
|
||||
"pyarrow",
|
||||
) from exc
|
||||
raise
|
||||
|
||||
config = ingest_config.get_method_config("parquet")
|
||||
config.update(kwargs)
|
||||
try:
|
||||
ingestor = ParquetIngestor(**config)
|
||||
except ImportError as exc:
|
||||
raise _missing_optional_dependency(
|
||||
"Parquet ingestion",
|
||||
"pyarrow",
|
||||
) from exc
|
||||
|
||||
def _run_single(path: Union[str, Path]) -> Union[ParquetData, Dict[str, Any]]:
|
||||
source_path = Path(path)
|
||||
|
||||
if method == "schema":
|
||||
return ingestor.extract_schema(source_path, **kwargs)
|
||||
if method == "metadata":
|
||||
return ingestor.extract_metadata(source_path, **kwargs)
|
||||
if method == "directory" or source_path.is_dir():
|
||||
return ingestor.ingest_directory(source_path, **kwargs)
|
||||
|
||||
return ingestor.ingest_file(source_path, **kwargs)
|
||||
|
||||
if isinstance(source, list):
|
||||
return [_run_single(path) for path in source]
|
||||
|
||||
return _run_single(source)
|
||||
|
||||
except ConfigurationError:
|
||||
raise
|
||||
except Exception as e:
|
||||
logger.error(f"Failed to ingest Parquet: {e}")
|
||||
raise
|
||||
|
||||
|
||||
def ingest_web(
|
||||
source: Union[str, List[str]], method: str = "url", **kwargs
|
||||
) -> Union[WebContent, List[WebContent], Dict[str, Any]]:
|
||||
@@ -450,7 +553,8 @@ def ingest_repository(
|
||||
"""
|
||||
Ingest repository from source (convenience function).
|
||||
|
||||
This is a user-friendly wrapper that ingests repositories using the specified method.
|
||||
This is a user-friendly wrapper that ingests repositories using the
|
||||
specified method.
|
||||
|
||||
Args:
|
||||
source: Repository URL or local path
|
||||
@@ -465,7 +569,9 @@ def ingest_repository(
|
||||
|
||||
Examples:
|
||||
>>> from semantica.ingest.methods import ingest_repository
|
||||
>>> repo_data = ingest_repository("https://github.com/user/repo.git", method="git")
|
||||
>>> repo_data = ingest_repository(
|
||||
... "https://github.com/user/repo.git", method="git"
|
||||
... )
|
||||
>>> analysis = ingest_repository("./repo", method="analyze")
|
||||
"""
|
||||
# Check for custom method in registry
|
||||
@@ -483,7 +589,9 @@ def ingest_repository(
|
||||
from .repo_ingestor import RepoIngestor
|
||||
except ModuleNotFoundError as exc:
|
||||
if _is_missing_dependency(exc, "git"):
|
||||
raise _missing_optional_dependency("Repository ingestion", "GitPython") from exc
|
||||
raise _missing_optional_dependency(
|
||||
"Repository ingestion", "GitPython"
|
||||
) from exc
|
||||
raise
|
||||
|
||||
# Get config
|
||||
@@ -634,19 +742,19 @@ def ingest_ontology(
|
||||
ingestor = OntologyIngestor(**config)
|
||||
|
||||
source_path = str(source) if isinstance(source, (str, Path)) else None
|
||||
|
||||
|
||||
if method == "file" and source_path:
|
||||
if isinstance(source, list):
|
||||
return [ingestor.ingest_ontology(str(s), **kwargs) for s in source]
|
||||
return ingestor.ingest_ontology(source_path, **kwargs)
|
||||
elif method == "directory" and source_path:
|
||||
recursive = kwargs.get("recursive", ingest_config.get("recursive", True))
|
||||
return ingestor.ingest_directory(source_path, recursive=recursive, **kwargs)
|
||||
recursive = kwargs.get("recursive", ingest_config.get("recursive", True))
|
||||
return ingestor.ingest_directory(source_path, recursive=recursive, **kwargs)
|
||||
else:
|
||||
# Default: try as file
|
||||
if isinstance(source, list):
|
||||
# Default: try as file
|
||||
if isinstance(source, list):
|
||||
return [ingestor.ingest_ontology(str(s), **kwargs) for s in source]
|
||||
return ingestor.ingest_ontology(str(source), **kwargs)
|
||||
return ingestor.ingest_ontology(str(source), **kwargs)
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Failed to ingest ontology: {e}")
|
||||
@@ -743,7 +851,8 @@ def ingest_mcp(
|
||||
the specified method. Works with Python and FastMCP MCP servers.
|
||||
|
||||
Args:
|
||||
source: MCP server URL (str) or configuration dict with "url" key, or server name (str) if already connected
|
||||
source: MCP server URL, configuration dict with "url" key, or server
|
||||
name if already connected
|
||||
- URL string: "http://localhost:8000/mcp"
|
||||
- Dict: {"url": "http://localhost:8000/mcp", "headers": {...}}
|
||||
method: Ingestion method (default: "resources")
|
||||
@@ -766,13 +875,25 @@ def ingest_mcp(
|
||||
>>> data = ingest_mcp("http://localhost:8000/mcp", method="resources")
|
||||
>>> # Connect via URL dict and ingest all resources
|
||||
>>> data = ingest_mcp(
|
||||
... {"url": "https://api.example.com/mcp", "headers": {"Authorization": "Bearer token"}},
|
||||
... {
|
||||
... "url": "https://api.example.com/mcp",
|
||||
... "headers": {"Authorization": "Bearer token"},
|
||||
... },
|
||||
... method="all"
|
||||
... )
|
||||
>>> # Ingest from already connected server
|
||||
>>> data = ingest_mcp("server1", method="resources", resource_uris=["resource://example"])
|
||||
>>> data = ingest_mcp(
|
||||
... "server1",
|
||||
... method="resources",
|
||||
... resource_uris=["resource://example"],
|
||||
... )
|
||||
>>> # Call tool
|
||||
>>> result = ingest_mcp("server1", method="tools", tool_name="get_data", tool_arguments={})
|
||||
>>> result = ingest_mcp(
|
||||
... "server1",
|
||||
... method="tools",
|
||||
... tool_name="get_data",
|
||||
... tool_arguments={},
|
||||
... )
|
||||
"""
|
||||
# Check for custom method in registry
|
||||
custom_method = method_registry.get("mcp", method)
|
||||
@@ -838,7 +959,8 @@ def ingest_mcp(
|
||||
)
|
||||
else:
|
||||
raise ProcessingError(
|
||||
"Source must be MCP server URL (str), configuration dict with 'url' key, "
|
||||
"Source must be MCP server URL (str), configuration dict "
|
||||
"with 'url' key, "
|
||||
"or server name (str) if already connected"
|
||||
)
|
||||
|
||||
@@ -890,6 +1012,7 @@ def ingest(
|
||||
- "email": Email ingestion
|
||||
- "db": Database ingestion
|
||||
- "ontology": Ontology ingestion
|
||||
- "parquet": Apache Parquet file or directory ingestion
|
||||
method: Optional specific ingestion method
|
||||
**kwargs: Additional options passed to ingestor
|
||||
|
||||
@@ -909,24 +1032,40 @@ def ingest(
|
||||
if not source_type:
|
||||
if isinstance(sources, (str, Path)):
|
||||
source_str = str(sources)
|
||||
if source_str.startswith(("http://", "https://")):
|
||||
source_str_lower = source_str.lower()
|
||||
if source_str_lower.startswith(("http://", "https://")):
|
||||
# Check if it's a feed URL
|
||||
if any(ext in source_str for ext in [".xml", "/feed", "/rss", "/atom"]):
|
||||
if any(
|
||||
ext in source_str_lower
|
||||
for ext in [".xml", "/feed", "/rss", "/atom"]
|
||||
):
|
||||
source_type = "feed"
|
||||
else:
|
||||
source_type = "web"
|
||||
elif source_str.startswith(
|
||||
elif source_str_lower.startswith(
|
||||
("postgresql://", "mysql://", "sqlite://", "oracle://", "mssql://")
|
||||
):
|
||||
source_type = "db"
|
||||
elif source_str.startswith(
|
||||
("git@", "https://github.com", "https://gitlab.com")
|
||||
elif source_str.startswith("git@") or source_str_lower.startswith(
|
||||
("https://github.com", "https://gitlab.com")
|
||||
):
|
||||
source_type = "repo"
|
||||
elif source_str.endswith((".ttl", ".owl", ".rdf", ".jsonld", ".n3", ".nt")):
|
||||
elif source_str_lower.endswith(
|
||||
(".ttl", ".owl", ".rdf", ".jsonld", ".n3", ".nt")
|
||||
):
|
||||
source_type = "ontology"
|
||||
elif source_str_lower.endswith((".parquet", ".pq")):
|
||||
source_type = "parquet"
|
||||
else:
|
||||
source_type = "file"
|
||||
elif (
|
||||
isinstance(sources, list)
|
||||
and sources
|
||||
and all(
|
||||
str(source).lower().endswith((".parquet", ".pq")) for source in sources
|
||||
)
|
||||
):
|
||||
source_type = "parquet"
|
||||
else:
|
||||
source_type = "file"
|
||||
|
||||
@@ -953,6 +1092,8 @@ def ingest(
|
||||
raise ProcessingError("Email ingestion requires configuration dictionary")
|
||||
elif source_type == "db":
|
||||
return {"data": ingest_database(sources, method=method, **kwargs)}
|
||||
elif source_type == "parquet":
|
||||
return {"data": ingest_parquet(sources, method=method or "file", **kwargs)}
|
||||
elif source_type == "ontology":
|
||||
return {"ontology": ingest_ontology(sources, method=method or "file", **kwargs)}
|
||||
elif source_type == "mcp":
|
||||
@@ -966,7 +1107,8 @@ def get_ingest_method(task: str, name: str) -> Optional[Callable]:
|
||||
Get a registered ingestion method.
|
||||
|
||||
Args:
|
||||
task: Task type ("file", "web", "feed", "stream", "repo", "email", "db", "mcp", "ingest")
|
||||
task: Task type ("file", "web", "feed", "stream", "repo", "email",
|
||||
"db", "mcp", "ingest")
|
||||
name: Method name
|
||||
|
||||
Returns:
|
||||
@@ -1030,6 +1172,12 @@ method_registry.register("db", "mysql", ingest_database)
|
||||
method_registry.register("db", "sqlite", ingest_database)
|
||||
method_registry.register("db", "oracle", ingest_database)
|
||||
method_registry.register("db", "mssql", ingest_database)
|
||||
method_registry.register("parquet", "default", ingest_parquet)
|
||||
method_registry.register("parquet", "file", ingest_parquet)
|
||||
method_registry.register("parquet", "directory", ingest_parquet)
|
||||
method_registry.register("parquet", "schema", ingest_parquet)
|
||||
method_registry.register("parquet", "metadata", ingest_parquet)
|
||||
method_registry.register("file", "parquet", ingest_parquet)
|
||||
method_registry.register("mcp", "default", ingest_mcp)
|
||||
method_registry.register("mcp", "resources", ingest_mcp)
|
||||
method_registry.register("mcp", "tools", ingest_mcp)
|
||||
|
||||
@@ -0,0 +1,766 @@
|
||||
"""
|
||||
Apache Parquet Ingestion Module
|
||||
|
||||
This module provides dedicated Parquet ingestion for local files and partitioned
|
||||
directories. It uses PyArrow when available so callers can read selected
|
||||
columns, inspect schemas and file metadata, and ingest Hive-style partitioned
|
||||
datasets without database credentials.
|
||||
|
||||
Example Usage:
|
||||
>>> from semantica.ingest import ParquetIngestor
|
||||
>>> ingestor = ParquetIngestor()
|
||||
>>> data = ingestor.ingest_file("events.parquet", columns=["id", "event_type"])
|
||||
>>> schema = ingestor.extract_schema("events.parquet")
|
||||
>>> partitioned = ingestor.ingest_directory("./events_by_date")
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from dataclasses import dataclass, field
|
||||
from datetime import datetime
|
||||
from pathlib import Path
|
||||
from typing import Any, Dict, Iterable, List, Optional, Sequence, Union
|
||||
|
||||
try:
|
||||
import pyarrow as pa
|
||||
import pyarrow.dataset as ds
|
||||
import pyarrow.parquet as pq
|
||||
|
||||
PARQUET_AVAILABLE = True
|
||||
except (ImportError, OSError):
|
||||
pa = None
|
||||
ds = None
|
||||
pq = None
|
||||
PARQUET_AVAILABLE = False
|
||||
|
||||
from ..utils.exceptions import ProcessingError, ValidationError
|
||||
from ..utils.logging import get_logger
|
||||
from ..utils.progress_tracker import get_progress_tracker
|
||||
|
||||
|
||||
@dataclass
|
||||
class ParquetData:
|
||||
"""Parquet ingestion result."""
|
||||
|
||||
data: List[Dict[str, Any]]
|
||||
row_count: int
|
||||
columns: List[str]
|
||||
schema: Dict[str, Any]
|
||||
source: str
|
||||
metadata: Dict[str, Any] = field(default_factory=dict)
|
||||
ingested_at: datetime = field(default_factory=datetime.now)
|
||||
|
||||
|
||||
class ParquetIngestor:
|
||||
"""
|
||||
Dedicated Parquet ingestion handler.
|
||||
|
||||
Features:
|
||||
- Single Parquet file ingestion
|
||||
- Partitioned directory ingestion with Hive partition discovery
|
||||
- Selective column reads
|
||||
- Schema and file metadata extraction
|
||||
- Optional row limits for sampling large files
|
||||
"""
|
||||
|
||||
def __init__(self, config: Optional[Dict[str, Any]] = None, **kwargs):
|
||||
"""
|
||||
Initialize Parquet ingestor.
|
||||
|
||||
Args:
|
||||
config: Optional configuration dictionary
|
||||
**kwargs: Additional configuration options
|
||||
|
||||
Raises:
|
||||
ImportError: If pyarrow is not installed
|
||||
"""
|
||||
if not PARQUET_AVAILABLE:
|
||||
raise ImportError(
|
||||
"pyarrow is required for ParquetIngestor. "
|
||||
"Install it with: pip install pyarrow"
|
||||
)
|
||||
|
||||
self.logger = get_logger("parquet_ingestor")
|
||||
self.config = config or {}
|
||||
self.config.update(kwargs)
|
||||
self.progress_tracker = get_progress_tracker()
|
||||
if not self.progress_tracker.enabled:
|
||||
self.progress_tracker.enabled = True
|
||||
|
||||
self.logger.debug("Parquet ingestor initialized")
|
||||
|
||||
def ingest(
|
||||
self,
|
||||
source: Union[str, Path],
|
||||
columns: Optional[Union[str, Sequence[str]]] = None,
|
||||
limit: Optional[int] = None,
|
||||
filters: Any = None,
|
||||
include_data: bool = True,
|
||||
**options,
|
||||
) -> ParquetData:
|
||||
"""
|
||||
Ingest a Parquet file or partitioned Parquet directory.
|
||||
|
||||
Args:
|
||||
source: Parquet file or directory path
|
||||
columns: Optional column name or names to read
|
||||
limit: Optional maximum number of rows to return
|
||||
filters: Optional PyArrow filter expression or tuple filters
|
||||
include_data: If False, return schema and metadata without rows
|
||||
**options: Additional options
|
||||
|
||||
Returns:
|
||||
ParquetData: Ingested data and metadata
|
||||
"""
|
||||
source_path = Path(source)
|
||||
if source_path.is_dir():
|
||||
return self.ingest_directory(
|
||||
source_path,
|
||||
columns=columns,
|
||||
limit=limit,
|
||||
filters=filters,
|
||||
include_data=include_data,
|
||||
**options,
|
||||
)
|
||||
return self.ingest_file(
|
||||
source_path,
|
||||
columns=columns,
|
||||
limit=limit,
|
||||
filters=filters,
|
||||
include_data=include_data,
|
||||
**options,
|
||||
)
|
||||
|
||||
def ingest_file(
|
||||
self,
|
||||
file_path: Union[str, Path],
|
||||
columns: Optional[Union[str, Sequence[str]]] = None,
|
||||
limit: Optional[int] = None,
|
||||
filters: Any = None,
|
||||
include_data: bool = True,
|
||||
batch_size: Optional[int] = None,
|
||||
**options,
|
||||
) -> ParquetData:
|
||||
"""
|
||||
Ingest a single Parquet file.
|
||||
|
||||
Args:
|
||||
file_path: Path to Parquet file
|
||||
columns: Optional column name or names to read
|
||||
limit: Optional maximum number of rows to return
|
||||
filters: Optional PyArrow-compatible filters
|
||||
include_data: If False, skip reading row data
|
||||
batch_size: Batch size used when sampling with limit
|
||||
**options: Additional options
|
||||
|
||||
Returns:
|
||||
ParquetData: Ingested data, schema, and metadata
|
||||
"""
|
||||
file_path = Path(file_path)
|
||||
self._validate_file(file_path)
|
||||
|
||||
tracking_id = self.progress_tracker.start_tracking(
|
||||
file=str(file_path),
|
||||
module="ingest",
|
||||
submodule="ParquetIngestor",
|
||||
message=f"Ingesting Parquet: {file_path.name}",
|
||||
)
|
||||
|
||||
try:
|
||||
parquet_file = pq.ParquetFile(str(file_path))
|
||||
selected_columns = self._normalize_columns(
|
||||
columns,
|
||||
[field.name for field in parquet_file.schema_arrow],
|
||||
)
|
||||
metadata = self._file_metadata(file_path, parquet_file)
|
||||
|
||||
if include_data:
|
||||
table = self._read_file_table(
|
||||
file_path=file_path,
|
||||
parquet_file=parquet_file,
|
||||
columns=selected_columns,
|
||||
limit=limit,
|
||||
filters=filters,
|
||||
batch_size=batch_size,
|
||||
)
|
||||
data = table.to_pylist()
|
||||
schema = self._schema_to_dict(table.schema)
|
||||
result_columns = list(table.column_names)
|
||||
else:
|
||||
selected_schema = self._select_schema(
|
||||
parquet_file.schema_arrow, selected_columns
|
||||
)
|
||||
data = []
|
||||
schema = self._schema_to_dict(selected_schema)
|
||||
result_columns = [field.name for field in selected_schema]
|
||||
|
||||
metadata.update(
|
||||
{
|
||||
"returned_rows": len(data),
|
||||
"selected_columns": result_columns,
|
||||
"filters_applied": filters is not None,
|
||||
"limit": limit,
|
||||
"include_data": include_data,
|
||||
}
|
||||
)
|
||||
|
||||
self.progress_tracker.stop_tracking(
|
||||
tracking_id,
|
||||
status="completed",
|
||||
message=f"Ingested Parquet: {len(data)} rows",
|
||||
)
|
||||
|
||||
self.logger.info(
|
||||
f"Parquet ingestion completed: {len(data)} row(s) from {file_path}"
|
||||
)
|
||||
|
||||
return ParquetData(
|
||||
data=data,
|
||||
row_count=len(data),
|
||||
columns=result_columns,
|
||||
schema=schema,
|
||||
source=str(file_path),
|
||||
metadata=metadata,
|
||||
)
|
||||
|
||||
except (ValidationError, ProcessingError):
|
||||
self.progress_tracker.stop_tracking(
|
||||
tracking_id, status="failed", message="Parquet ingestion failed"
|
||||
)
|
||||
raise
|
||||
except Exception as e:
|
||||
self.progress_tracker.stop_tracking(
|
||||
tracking_id, status="failed", message=str(e)
|
||||
)
|
||||
self.logger.error(f"Failed to ingest Parquet {file_path}: {e}")
|
||||
raise ProcessingError(f"Failed to ingest Parquet: {e}") from e
|
||||
|
||||
def ingest_directory(
|
||||
self,
|
||||
directory_path: Union[str, Path],
|
||||
columns: Optional[Union[str, Sequence[str]]] = None,
|
||||
limit: Optional[int] = None,
|
||||
filters: Any = None,
|
||||
include_data: bool = True,
|
||||
partitioning: Optional[Union[str, Any]] = "hive",
|
||||
**options,
|
||||
) -> ParquetData:
|
||||
"""
|
||||
Ingest a directory containing Parquet files.
|
||||
|
||||
Hive-style partitions such as ``country=US/year=2026`` are discovered
|
||||
by default and included as partition columns in the returned schema/data.
|
||||
|
||||
Args:
|
||||
directory_path: Directory containing Parquet files
|
||||
columns: Optional column name or names to read
|
||||
limit: Optional maximum number of rows to return
|
||||
filters: Optional PyArrow filter expression or tuple filters
|
||||
include_data: If False, return only schema and metadata
|
||||
partitioning: PyArrow partitioning mode, defaults to "hive"
|
||||
**options: Additional options
|
||||
|
||||
Returns:
|
||||
ParquetData: Ingested dataset data and metadata
|
||||
"""
|
||||
directory_path = Path(directory_path)
|
||||
parquet_files = self._validate_directory(directory_path)
|
||||
if limit is not None and limit < 0:
|
||||
raise ValidationError("limit must be greater than or equal to 0")
|
||||
|
||||
tracking_id = self.progress_tracker.start_tracking(
|
||||
file=str(directory_path),
|
||||
module="ingest",
|
||||
submodule="ParquetIngestor",
|
||||
message=f"Ingesting Parquet directory: {directory_path.name}",
|
||||
)
|
||||
|
||||
try:
|
||||
dataset = ds.dataset(
|
||||
str(directory_path),
|
||||
format="parquet",
|
||||
partitioning=partitioning,
|
||||
)
|
||||
selected_columns = self._normalize_columns(
|
||||
columns,
|
||||
[field.name for field in dataset.schema],
|
||||
)
|
||||
filter_expression = self._dataset_filter(filters)
|
||||
metadata = self._directory_metadata(
|
||||
directory_path,
|
||||
parquet_files,
|
||||
partitioning=partitioning,
|
||||
)
|
||||
|
||||
if include_data:
|
||||
if limit is not None:
|
||||
table = dataset.head(
|
||||
limit,
|
||||
columns=selected_columns,
|
||||
filter=filter_expression,
|
||||
)
|
||||
else:
|
||||
table = dataset.to_table(
|
||||
columns=selected_columns,
|
||||
filter=filter_expression,
|
||||
)
|
||||
data = table.to_pylist()
|
||||
schema = self._schema_to_dict(table.schema)
|
||||
result_columns = list(table.column_names)
|
||||
else:
|
||||
selected_schema = self._select_schema(dataset.schema, selected_columns)
|
||||
data = []
|
||||
schema = self._schema_to_dict(selected_schema)
|
||||
result_columns = [field.name for field in selected_schema]
|
||||
|
||||
metadata.update(
|
||||
{
|
||||
"returned_rows": len(data),
|
||||
"selected_columns": result_columns,
|
||||
"filters_applied": filters is not None,
|
||||
"limit": limit,
|
||||
"include_data": include_data,
|
||||
}
|
||||
)
|
||||
|
||||
self.progress_tracker.stop_tracking(
|
||||
tracking_id,
|
||||
status="completed",
|
||||
message=f"Ingested Parquet directory: {len(data)} rows",
|
||||
)
|
||||
|
||||
self.logger.info(
|
||||
"Parquet directory ingestion completed: "
|
||||
f"{len(data)} row(s) from {directory_path}"
|
||||
)
|
||||
|
||||
return ParquetData(
|
||||
data=data,
|
||||
row_count=len(data),
|
||||
columns=result_columns,
|
||||
schema=schema,
|
||||
source=str(directory_path),
|
||||
metadata=metadata,
|
||||
)
|
||||
|
||||
except (ValidationError, ProcessingError):
|
||||
self.progress_tracker.stop_tracking(
|
||||
tracking_id,
|
||||
status="failed",
|
||||
message="Parquet directory ingestion failed",
|
||||
)
|
||||
raise
|
||||
except Exception as e:
|
||||
self.progress_tracker.stop_tracking(
|
||||
tracking_id, status="failed", message=str(e)
|
||||
)
|
||||
self.logger.error(
|
||||
f"Failed to ingest Parquet directory {directory_path}: {e}"
|
||||
)
|
||||
raise ProcessingError(f"Failed to ingest Parquet directory: {e}") from e
|
||||
|
||||
def read_columns(
|
||||
self,
|
||||
source: Union[str, Path],
|
||||
columns: Union[str, Sequence[str]],
|
||||
**options,
|
||||
) -> ParquetData:
|
||||
"""
|
||||
Read selected columns from a Parquet file or directory.
|
||||
|
||||
Args:
|
||||
source: Parquet file or directory path
|
||||
columns: Column name or names to read
|
||||
**options: Additional ingestion options
|
||||
|
||||
Returns:
|
||||
ParquetData: Ingested data containing only selected columns
|
||||
"""
|
||||
return self.ingest(source, columns=columns, **options)
|
||||
|
||||
def extract_schema(self, source: Union[str, Path], **options) -> Dict[str, Any]:
|
||||
"""
|
||||
Extract schema from a Parquet file or directory.
|
||||
|
||||
Args:
|
||||
source: Parquet file or directory path
|
||||
**options: Additional options
|
||||
|
||||
Returns:
|
||||
dict: Schema with column names, types, nullability, and metadata
|
||||
"""
|
||||
source_path = Path(source)
|
||||
if source_path.is_dir():
|
||||
self._validate_directory(source_path)
|
||||
dataset = ds.dataset(
|
||||
str(source_path),
|
||||
format="parquet",
|
||||
partitioning=options.get("partitioning", "hive"),
|
||||
)
|
||||
return self._schema_to_dict(dataset.schema)
|
||||
|
||||
self._validate_file(source_path)
|
||||
parquet_file = pq.ParquetFile(str(source_path))
|
||||
return self._schema_to_dict(parquet_file.schema_arrow)
|
||||
|
||||
def extract_metadata(self, source: Union[str, Path], **options) -> Dict[str, Any]:
|
||||
"""
|
||||
Extract Parquet file or directory metadata without reading row data.
|
||||
|
||||
Args:
|
||||
source: Parquet file or directory path
|
||||
**options: Additional options
|
||||
|
||||
Returns:
|
||||
dict: Row counts, row groups, compression, partitions, and file info
|
||||
"""
|
||||
source_path = Path(source)
|
||||
if source_path.is_dir():
|
||||
parquet_files = self._validate_directory(source_path)
|
||||
return self._directory_metadata(
|
||||
source_path,
|
||||
parquet_files,
|
||||
partitioning=options.get("partitioning", "hive"),
|
||||
)
|
||||
|
||||
self._validate_file(source_path)
|
||||
parquet_file = pq.ParquetFile(str(source_path))
|
||||
return self._file_metadata(source_path, parquet_file)
|
||||
|
||||
def _read_file_table(
|
||||
self,
|
||||
file_path: Path,
|
||||
parquet_file: Any,
|
||||
columns: Optional[List[str]],
|
||||
limit: Optional[int],
|
||||
filters: Any,
|
||||
batch_size: Optional[int],
|
||||
) -> Any:
|
||||
"""Read a Parquet file, using batches when a simple limit is requested."""
|
||||
if limit is not None and limit < 0:
|
||||
raise ValidationError("limit must be greater than or equal to 0")
|
||||
|
||||
if limit == 0:
|
||||
return pa.Table.from_batches(
|
||||
[],
|
||||
schema=self._select_schema(parquet_file.schema_arrow, columns),
|
||||
)
|
||||
|
||||
if limit is not None and filters is None:
|
||||
return self._read_file_limited(parquet_file, columns, limit, batch_size)
|
||||
|
||||
table = pq.read_table(str(file_path), columns=columns, filters=filters)
|
||||
if limit is not None:
|
||||
table = table.slice(0, limit)
|
||||
return table
|
||||
|
||||
def _read_file_limited(
|
||||
self,
|
||||
parquet_file: Any,
|
||||
columns: Optional[List[str]],
|
||||
limit: int,
|
||||
batch_size: Optional[int],
|
||||
) -> Any:
|
||||
"""Read at most ``limit`` rows from a file without loading the full file."""
|
||||
batches = []
|
||||
remaining = limit
|
||||
effective_batch_size = batch_size or min(max(limit, 1), 65_536)
|
||||
|
||||
for batch in parquet_file.iter_batches(
|
||||
batch_size=effective_batch_size,
|
||||
columns=columns,
|
||||
):
|
||||
if batch.num_rows > remaining:
|
||||
batch = batch.slice(0, remaining)
|
||||
batches.append(batch)
|
||||
remaining -= batch.num_rows
|
||||
if remaining <= 0:
|
||||
break
|
||||
|
||||
return pa.Table.from_batches(
|
||||
batches,
|
||||
schema=self._select_schema(parquet_file.schema_arrow, columns),
|
||||
)
|
||||
|
||||
def _validate_file(self, file_path: Path) -> None:
|
||||
"""Validate a local Parquet file path."""
|
||||
if not file_path.exists():
|
||||
raise ValidationError(f"Parquet file not found: {file_path}")
|
||||
if not file_path.is_file():
|
||||
raise ValidationError(f"Path is not a file: {file_path}")
|
||||
if file_path.suffix.lower() not in {".parquet", ".pq"}:
|
||||
raise ValidationError(f"File is not a Parquet file: {file_path}")
|
||||
|
||||
def _validate_directory(self, directory_path: Path) -> List[Path]:
|
||||
"""Validate a Parquet directory and return contained Parquet files."""
|
||||
if not directory_path.exists():
|
||||
raise ValidationError(f"Parquet directory not found: {directory_path}")
|
||||
if not directory_path.is_dir():
|
||||
raise ValidationError(f"Path is not a directory: {directory_path}")
|
||||
|
||||
parquet_files = self._parquet_files(directory_path)
|
||||
if not parquet_files:
|
||||
raise ValidationError(
|
||||
f"No Parquet files found in directory: {directory_path}"
|
||||
)
|
||||
return parquet_files
|
||||
|
||||
def _parquet_files(self, directory_path: Path) -> List[Path]:
|
||||
"""Return Parquet files under a directory."""
|
||||
return sorted(
|
||||
path
|
||||
for path in directory_path.rglob("*")
|
||||
if path.is_file() and path.suffix.lower() in {".parquet", ".pq"}
|
||||
)
|
||||
|
||||
def _normalize_columns(
|
||||
self,
|
||||
columns: Optional[Union[str, Sequence[str]]],
|
||||
available_columns: Sequence[str],
|
||||
) -> Optional[List[str]]:
|
||||
"""Normalize and validate optional selected columns."""
|
||||
if columns is None:
|
||||
configured_columns = self.config.get("columns")
|
||||
if configured_columns is None:
|
||||
return None
|
||||
columns = configured_columns
|
||||
|
||||
if isinstance(columns, str):
|
||||
normalized = [columns]
|
||||
else:
|
||||
normalized = list(columns)
|
||||
|
||||
missing = [column for column in normalized if column not in available_columns]
|
||||
if missing:
|
||||
raise ValidationError(
|
||||
"Column(s) not found in Parquet schema: "
|
||||
f"{', '.join(missing)}. Available columns: "
|
||||
f"{', '.join(available_columns)}"
|
||||
)
|
||||
return normalized
|
||||
|
||||
def _select_schema(self, schema: Any, columns: Optional[List[str]]) -> Any:
|
||||
"""Return schema limited to selected columns when provided."""
|
||||
if columns is None:
|
||||
return schema
|
||||
fields = [schema.field(column) for column in columns]
|
||||
return pa.schema(fields, metadata=schema.metadata)
|
||||
|
||||
def _schema_to_dict(self, schema: Any) -> Dict[str, Any]:
|
||||
"""Convert PyArrow schema to serializable metadata."""
|
||||
fields = []
|
||||
for schema_field in schema:
|
||||
fields.append(
|
||||
{
|
||||
"name": schema_field.name,
|
||||
"type": str(schema_field.type),
|
||||
"nullable": schema_field.nullable,
|
||||
"metadata": self._decode_metadata_map(schema_field.metadata),
|
||||
}
|
||||
)
|
||||
|
||||
return {
|
||||
"columns": [field_info["name"] for field_info in fields],
|
||||
"fields": fields,
|
||||
"metadata": self._decode_metadata_map(schema.metadata),
|
||||
}
|
||||
|
||||
def _file_metadata(self, file_path: Path, parquet_file: Any) -> Dict[str, Any]:
|
||||
"""Extract metadata for a single Parquet file."""
|
||||
metadata = parquet_file.metadata
|
||||
compression_by_column = self._compression_by_column(metadata)
|
||||
|
||||
return {
|
||||
"format": "parquet",
|
||||
"source_type": "file",
|
||||
"file": str(file_path),
|
||||
"file_size": file_path.stat().st_size,
|
||||
"total_rows": metadata.num_rows,
|
||||
"row_groups": metadata.num_row_groups,
|
||||
"created_by": metadata.created_by,
|
||||
"format_version": getattr(metadata, "format_version", None),
|
||||
"serialized_size": getattr(metadata, "serialized_size", None),
|
||||
"schema_metadata": self._decode_metadata_map(metadata.metadata),
|
||||
"compression": {
|
||||
column: sorted(codecs)
|
||||
for column, codecs in compression_by_column.items()
|
||||
},
|
||||
"compression_codecs": sorted(
|
||||
{codec for codecs in compression_by_column.values() for codec in codecs}
|
||||
),
|
||||
}
|
||||
|
||||
def _directory_metadata(
|
||||
self,
|
||||
directory_path: Path,
|
||||
parquet_files: Sequence[Path],
|
||||
partitioning: Optional[Union[str, Any]],
|
||||
) -> Dict[str, Any]:
|
||||
"""Extract aggregate metadata for a Parquet directory."""
|
||||
file_entries = []
|
||||
total_rows = 0
|
||||
total_row_groups = 0
|
||||
compression_by_column: Dict[str, set] = {}
|
||||
partition_columns = set()
|
||||
partition_values: Dict[str, set] = {}
|
||||
|
||||
for parquet_path in parquet_files:
|
||||
parquet_file = pq.ParquetFile(str(parquet_path))
|
||||
file_metadata = self._file_metadata(parquet_path, parquet_file)
|
||||
partitions = self._partition_values(directory_path, parquet_path)
|
||||
|
||||
total_rows += file_metadata["total_rows"]
|
||||
total_row_groups += file_metadata["row_groups"]
|
||||
for column, codecs in file_metadata["compression"].items():
|
||||
compression_by_column.setdefault(column, set()).update(codecs)
|
||||
|
||||
for key, value in partitions.items():
|
||||
partition_columns.add(key)
|
||||
partition_values.setdefault(key, set()).add(value)
|
||||
|
||||
file_entries.append(
|
||||
{
|
||||
"path": str(parquet_path),
|
||||
"relative_path": str(parquet_path.relative_to(directory_path)),
|
||||
"rows": file_metadata["total_rows"],
|
||||
"row_groups": file_metadata["row_groups"],
|
||||
"file_size": file_metadata["file_size"],
|
||||
"partitions": partitions,
|
||||
}
|
||||
)
|
||||
|
||||
return {
|
||||
"format": "parquet",
|
||||
"source_type": "directory",
|
||||
"directory": str(directory_path),
|
||||
"file_count": len(parquet_files),
|
||||
"files": file_entries,
|
||||
"total_rows": total_rows,
|
||||
"row_groups": total_row_groups,
|
||||
"partitioning": partitioning,
|
||||
"partition_columns": sorted(partition_columns),
|
||||
"partition_values": {
|
||||
key: sorted(values) for key, values in partition_values.items()
|
||||
},
|
||||
"compression": {
|
||||
column: sorted(codecs)
|
||||
for column, codecs in compression_by_column.items()
|
||||
},
|
||||
"compression_codecs": sorted(
|
||||
{codec for codecs in compression_by_column.values() for codec in codecs}
|
||||
),
|
||||
}
|
||||
|
||||
def _compression_by_column(self, metadata: Any) -> Dict[str, set]:
|
||||
"""Return compression codecs used for each column across row groups."""
|
||||
compression_by_column: Dict[str, set] = {}
|
||||
for row_group_index in range(metadata.num_row_groups):
|
||||
row_group = metadata.row_group(row_group_index)
|
||||
for column_index in range(row_group.num_columns):
|
||||
column_chunk = row_group.column(column_index)
|
||||
column_name = column_chunk.path_in_schema
|
||||
compression = str(column_chunk.compression)
|
||||
compression_by_column.setdefault(column_name, set()).add(compression)
|
||||
return compression_by_column
|
||||
|
||||
def _partition_values(self, root: Path, parquet_path: Path) -> Dict[str, str]:
|
||||
"""Extract Hive-style partition key/value pairs from a file path."""
|
||||
partitions = {}
|
||||
relative_parent = parquet_path.parent.relative_to(root)
|
||||
for part in relative_parent.parts:
|
||||
if "=" not in part:
|
||||
continue
|
||||
key, value = part.split("=", 1)
|
||||
if key:
|
||||
partitions[key] = value
|
||||
return partitions
|
||||
|
||||
def _decode_metadata_map(
|
||||
self, metadata: Optional[Dict[Any, Any]]
|
||||
) -> Dict[str, str]:
|
||||
"""Decode PyArrow metadata bytes to strings."""
|
||||
if not metadata:
|
||||
return {}
|
||||
|
||||
decoded = {}
|
||||
for key, value in metadata.items():
|
||||
decoded[self._decode_metadata_value(key)] = self._decode_metadata_value(
|
||||
value
|
||||
)
|
||||
return decoded
|
||||
|
||||
def _decode_metadata_value(self, value: Any) -> str:
|
||||
"""Decode a metadata key or value."""
|
||||
if isinstance(value, bytes):
|
||||
return value.decode("utf-8", errors="replace")
|
||||
return str(value)
|
||||
|
||||
def _dataset_filter(self, filters: Any) -> Any:
|
||||
"""Convert simple tuple filters to a PyArrow dataset expression."""
|
||||
if filters is None:
|
||||
return None
|
||||
if self._is_filter_tuple(filters):
|
||||
return self._comparison_expression(*filters)
|
||||
if isinstance(filters, list):
|
||||
if all(self._is_filter_tuple(item) for item in filters):
|
||||
return self._and_expressions(
|
||||
self._comparison_expression(*item) for item in filters
|
||||
)
|
||||
if all(isinstance(group, list) for group in filters):
|
||||
return self._or_expressions(
|
||||
self._and_expressions(
|
||||
self._comparison_expression(*item) for item in group
|
||||
)
|
||||
for group in filters
|
||||
)
|
||||
return filters
|
||||
|
||||
def _is_filter_tuple(self, value: Any) -> bool:
|
||||
"""Return whether value is a simple (column, operator, value) filter."""
|
||||
return (
|
||||
isinstance(value, tuple)
|
||||
and len(value) == 3
|
||||
and isinstance(value[0], str)
|
||||
and isinstance(value[1], str)
|
||||
)
|
||||
|
||||
def _comparison_expression(self, column: str, operator: str, value: Any) -> Any:
|
||||
"""Create a PyArrow dataset comparison expression."""
|
||||
field = ds.field(column)
|
||||
if operator in {"=", "=="}:
|
||||
return field == value
|
||||
if operator == "!=":
|
||||
return field != value
|
||||
if operator == ">":
|
||||
return field > value
|
||||
if operator == ">=":
|
||||
return field >= value
|
||||
if operator == "<":
|
||||
return field < value
|
||||
if operator == "<=":
|
||||
return field <= value
|
||||
if operator.lower() == "in":
|
||||
return field.isin(value)
|
||||
if operator.lower() in {"not in", "not_in"}:
|
||||
return ~field.isin(value)
|
||||
raise ValidationError(f"Unsupported Parquet filter operator: {operator}")
|
||||
|
||||
def _and_expressions(self, expressions: Iterable[Any]) -> Any:
|
||||
"""Combine expressions with AND."""
|
||||
expression_list = list(expressions)
|
||||
if not expression_list:
|
||||
return None
|
||||
combined = expression_list[0]
|
||||
for expression in expression_list[1:]:
|
||||
combined = combined & expression
|
||||
return combined
|
||||
|
||||
def _or_expressions(self, expressions: Iterable[Any]) -> Any:
|
||||
"""Combine expressions with OR."""
|
||||
expression_list = [expr for expr in expressions if expr is not None]
|
||||
if not expression_list:
|
||||
return None
|
||||
combined = expression_list[0]
|
||||
for expression in expression_list[1:]:
|
||||
combined = combined | expression
|
||||
return combined
|
||||
@@ -13,6 +13,7 @@ Supported Registration Types:
|
||||
* "repo": Repository ingestion methods
|
||||
* "email": Email ingestion methods
|
||||
* "db": Database ingestion methods
|
||||
* "parquet": Parquet file and dataset ingestion methods
|
||||
* "ingest": General ingestion methods
|
||||
|
||||
Algorithms Used:
|
||||
@@ -24,7 +25,7 @@ Algorithms Used:
|
||||
|
||||
Key Features:
|
||||
- Method registry for custom ingestion methods
|
||||
- Task-based method organization (file, web, feed, stream, repo, email, db, ingest)
|
||||
- Task-based method organization by source category
|
||||
- Dynamic registration and unregistration
|
||||
- Easy discovery of available methods
|
||||
- Support for community-contributed extensions
|
||||
@@ -37,11 +38,13 @@ Global Instances:
|
||||
|
||||
Example Usage:
|
||||
>>> from semantica.ingest.registry import method_registry
|
||||
>>> method_registry.register("file", "custom_method", custom_file_ingestion_function)
|
||||
>>> method_registry.register(
|
||||
... "file", "custom_method", custom_file_ingestion_function
|
||||
... )
|
||||
>>> available = method_registry.list_all("file")
|
||||
"""
|
||||
|
||||
from typing import Any, Callable, Dict, List, Optional
|
||||
from typing import Callable, Dict, List, Optional
|
||||
|
||||
|
||||
class MethodRegistry:
|
||||
@@ -56,6 +59,7 @@ class MethodRegistry:
|
||||
"email": {},
|
||||
"db": {},
|
||||
"mcp": {},
|
||||
"parquet": {},
|
||||
"ingest": {},
|
||||
}
|
||||
|
||||
@@ -65,7 +69,8 @@ class MethodRegistry:
|
||||
Register a custom ingestion method.
|
||||
|
||||
Args:
|
||||
task: Task type ("file", "web", "feed", "stream", "repo", "email", "db", "mcp", "ingest")
|
||||
task: Task type such as "file", "web", "feed", "stream",
|
||||
"repo", "email", "db", "mcp", "parquet", or "ingest"
|
||||
name: Method name
|
||||
method_func: Method function
|
||||
"""
|
||||
@@ -79,7 +84,8 @@ class MethodRegistry:
|
||||
Get method by task and name.
|
||||
|
||||
Args:
|
||||
task: Task type ("file", "web", "feed", "stream", "repo", "email", "db", "mcp", "ingest")
|
||||
task: Task type such as "file", "web", "feed", "stream",
|
||||
"repo", "email", "db", "mcp", "parquet", or "ingest"
|
||||
name: Method name
|
||||
|
||||
Returns:
|
||||
@@ -108,7 +114,8 @@ class MethodRegistry:
|
||||
Unregister a method.
|
||||
|
||||
Args:
|
||||
task: Task type ("file", "web", "feed", "stream", "repo", "email", "db", "mcp", "ingest")
|
||||
task: Task type such as "file", "web", "feed", "stream",
|
||||
"repo", "email", "db", "mcp", "parquet", or "ingest"
|
||||
name: Method name
|
||||
"""
|
||||
if task in cls._methods and name in cls._methods[task]:
|
||||
|
||||
@@ -33,10 +33,10 @@ Example Usage:
|
||||
>>> from semantica.utils import SUPPORTED_DOCUMENT_FORMATS, DEFAULT_CONFIG
|
||||
>>> if file_extension in SUPPORTED_DOCUMENT_FORMATS:
|
||||
... process_document(file_path)
|
||||
>>>
|
||||
>>>
|
||||
>>> config = DEFAULT_CONFIG.copy()
|
||||
>>> config["processing"]["batch_size"] = 200
|
||||
>>>
|
||||
>>>
|
||||
>>> from semantica.utils import ERROR_CODES, PERFORMANCE_THRESHOLDS
|
||||
>>> error_code = ERROR_CODES["VALIDATION_ERROR"]
|
||||
>>> max_time = PERFORMANCE_THRESHOLDS["max_processing_time"]
|
||||
@@ -57,6 +57,8 @@ SUPPORTED_DOCUMENT_FORMATS = [
|
||||
"csv",
|
||||
"xlsx",
|
||||
"pptx",
|
||||
"parquet",
|
||||
"pq",
|
||||
]
|
||||
|
||||
SUPPORTED_IMAGE_FORMATS = ["jpg", "jpeg", "png", "gif", "bmp", "tiff", "webp", "svg"]
|
||||
|
||||
@@ -4,7 +4,6 @@ import sys
|
||||
import textwrap
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
REPO_ROOT = Path(__file__).resolve().parents[2]
|
||||
|
||||
|
||||
@@ -50,7 +49,7 @@ def test_file_ingestion_imports_without_optional_backends() -> None:
|
||||
from semantica.ingest import FileIngestor, ingest_file
|
||||
print(FileIngestor.__name__, callable(ingest_file))
|
||||
""",
|
||||
("git", "bs4"),
|
||||
("git", "bs4", "pyarrow"),
|
||||
)
|
||||
|
||||
assert result.returncode == 0, result.stderr
|
||||
@@ -76,3 +75,24 @@ else:
|
||||
assert "ConfigurationError" in result.stdout
|
||||
assert "Repository ingestion" in result.stdout
|
||||
assert "GitPython" in result.stdout
|
||||
|
||||
|
||||
def test_parquet_ingestion_reports_missing_pyarrow_when_used() -> None:
|
||||
result = _run_python_with_blocked_modules(
|
||||
"""
|
||||
from semantica.ingest import ingest_parquet
|
||||
|
||||
try:
|
||||
ingest_parquet("events.parquet")
|
||||
except Exception as exc:
|
||||
print(type(exc).__name__, exc)
|
||||
else:
|
||||
raise SystemExit("expected parquet ingestion to fail without pyarrow")
|
||||
""",
|
||||
("pyarrow",),
|
||||
)
|
||||
|
||||
assert result.returncode == 0, result.stderr
|
||||
assert "ConfigurationError" in result.stdout
|
||||
assert "Parquet ingestion" in result.stdout
|
||||
assert "pyarrow" in result.stdout
|
||||
|
||||
@@ -0,0 +1,150 @@
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
pa = pytest.importorskip("pyarrow")
|
||||
pq = pytest.importorskip("pyarrow.parquet")
|
||||
|
||||
from semantica.ingest import ( # noqa: E402
|
||||
ParquetData,
|
||||
ParquetIngestor,
|
||||
ingest,
|
||||
ingest_file,
|
||||
ingest_parquet,
|
||||
list_available_methods,
|
||||
)
|
||||
from semantica.ingest.file_ingestor import FileTypeDetector # noqa: E402
|
||||
from semantica.utils.exceptions import ValidationError # noqa: E402
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def sample_parquet(tmp_path: Path) -> Path:
|
||||
path = tmp_path / "events.parquet"
|
||||
table = pa.table(
|
||||
{
|
||||
"id": [1, 2, 3],
|
||||
"name": ["alpha", "beta", "gamma"],
|
||||
"score": [0.7, 0.8, 0.9],
|
||||
"city": ["Pune", "Delhi", "Mumbai"],
|
||||
}
|
||||
)
|
||||
pq.write_table(table, path, compression="snappy")
|
||||
return path
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def partitioned_parquet(tmp_path: Path) -> Path:
|
||||
root = tmp_path / "events_partitioned"
|
||||
|
||||
us_2025 = root / "country=US" / "year=2025"
|
||||
us_2025.mkdir(parents=True)
|
||||
pq.write_table(
|
||||
pa.table({"id": [1, 2], "value": ["a", "b"]}),
|
||||
us_2025 / "part-0.parquet",
|
||||
compression="gzip",
|
||||
)
|
||||
|
||||
ca_2026 = root / "country=CA" / "year=2026"
|
||||
ca_2026.mkdir(parents=True)
|
||||
pq.write_table(
|
||||
pa.table({"id": [3], "value": ["c"]}),
|
||||
ca_2026 / "part-1.parquet",
|
||||
compression="gzip",
|
||||
)
|
||||
|
||||
return root
|
||||
|
||||
|
||||
def test_parquet_file_ingestion_reads_data_schema_and_metadata(
|
||||
sample_parquet: Path,
|
||||
) -> None:
|
||||
ingestor = ParquetIngestor()
|
||||
|
||||
result = ingestor.ingest_file(sample_parquet)
|
||||
|
||||
assert isinstance(result, ParquetData)
|
||||
assert result.row_count == 3
|
||||
assert result.columns == ["id", "name", "score", "city"]
|
||||
assert result.data[0]["name"] == "alpha"
|
||||
assert result.schema["columns"] == ["id", "name", "score", "city"]
|
||||
assert result.schema["fields"][0]["type"] == "int64"
|
||||
assert result.metadata["total_rows"] == 3
|
||||
assert result.metadata["row_groups"] == 1
|
||||
assert result.metadata["compression_codecs"] == ["SNAPPY"]
|
||||
|
||||
|
||||
def test_parquet_selective_column_reading_with_limit(sample_parquet: Path) -> None:
|
||||
ingestor = ParquetIngestor()
|
||||
|
||||
result = ingestor.ingest_file(sample_parquet, columns=["id", "name"], limit=2)
|
||||
|
||||
assert result.row_count == 2
|
||||
assert result.columns == ["id", "name"]
|
||||
assert result.data == [{"id": 1, "name": "alpha"}, {"id": 2, "name": "beta"}]
|
||||
assert result.metadata["selected_columns"] == ["id", "name"]
|
||||
assert result.metadata["limit"] == 2
|
||||
|
||||
|
||||
def test_parquet_schema_and_metadata_can_be_extracted_without_rows(
|
||||
sample_parquet: Path,
|
||||
) -> None:
|
||||
ingestor = ParquetIngestor()
|
||||
|
||||
schema = ingestor.extract_schema(sample_parquet)
|
||||
metadata = ingestor.extract_metadata(sample_parquet)
|
||||
result = ingestor.ingest_file(sample_parquet, include_data=False)
|
||||
|
||||
assert schema["columns"] == ["id", "name", "score", "city"]
|
||||
assert metadata["total_rows"] == 3
|
||||
assert metadata["format"] == "parquet"
|
||||
assert result.row_count == 0
|
||||
assert result.data == []
|
||||
assert result.metadata["include_data"] is False
|
||||
|
||||
|
||||
def test_partitioned_parquet_directory_ingestion(partitioned_parquet: Path) -> None:
|
||||
ingestor = ParquetIngestor()
|
||||
|
||||
result = ingestor.ingest_directory(partitioned_parquet)
|
||||
|
||||
assert result.row_count == 3
|
||||
assert set(result.columns) == {"id", "value", "country", "year"}
|
||||
assert {row["country"] for row in result.data} == {"US", "CA"}
|
||||
assert result.metadata["file_count"] == 2
|
||||
assert result.metadata["total_rows"] == 3
|
||||
assert result.metadata["partition_columns"] == ["country", "year"]
|
||||
assert result.metadata["partition_values"] == {
|
||||
"country": ["CA", "US"],
|
||||
"year": ["2025", "2026"],
|
||||
}
|
||||
assert result.metadata["compression_codecs"] == ["GZIP"]
|
||||
|
||||
|
||||
def test_parquet_convenience_methods_and_unified_dispatch(sample_parquet: Path) -> None:
|
||||
direct = ingest_parquet(sample_parquet, columns=["name"])
|
||||
via_file_method = ingest_file(sample_parquet, method="parquet", limit=1)
|
||||
unified = ingest(sample_parquet)
|
||||
unified_batch = ingest([sample_parquet])
|
||||
methods = list_available_methods("parquet")
|
||||
|
||||
assert isinstance(direct, ParquetData)
|
||||
assert direct.columns == ["name"]
|
||||
assert isinstance(via_file_method, ParquetData)
|
||||
assert via_file_method.row_count == 1
|
||||
assert isinstance(unified["data"], ParquetData)
|
||||
assert isinstance(unified_batch["data"][0], ParquetData)
|
||||
assert "metadata" in methods["parquet"]
|
||||
|
||||
|
||||
def test_file_type_detector_recognizes_parquet_magic_number() -> None:
|
||||
detector = FileTypeDetector()
|
||||
|
||||
assert detector.detect_type("dataset", content=b"PAR1payload") == "parquet"
|
||||
assert detector.is_supported("parquet")
|
||||
|
||||
|
||||
def test_parquet_ingestion_rejects_negative_limit(sample_parquet: Path) -> None:
|
||||
ingestor = ParquetIngestor()
|
||||
|
||||
with pytest.raises(ValidationError):
|
||||
ingestor.ingest_file(sample_parquet, limit=-1)
|
||||
Reference in New Issue
Block a user