mirror of
https://github.com/semantica-agi/semantica.git
synced 2026-08-30 04:40:16 +00:00
Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
 KaifAhmad1
|
2ef6e9f4b1 | ||
|
Mohd Kaif
|
18da322e0d | ||
|
Luffy2208andKaifAhmad1
|
15d58f2b88 |
+27
-5
@@ -7,10 +7,27 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
|
||||
|
||||
---
|
||||
|
||||
## [Unreleased]
|
||||
## [0.5.0] - 2026-05-11
|
||||
|
||||
### Added
|
||||
|
||||
- **Distance Intelligence Embedding Cache Optimization** by @KaifAhmad1
|
||||
- Implemented per-session graph revision-based embedding cache to avoid re-scanning all nodes on every request
|
||||
- Added `get_cached_embeddings()` method to GraphSession with thread-safe caching and automatic invalidation
|
||||
- Updated distance matrix and semantic neighborhood endpoints to use cached embeddings for significant performance improvement
|
||||
- Added graph revision tracking using hash-based identifiers for cache invalidation
|
||||
- Implemented force refresh capability and automatic cache invalidation on graph modifications (add_nodes/add_edges)
|
||||
- Resolved TODO in `graph.py` for embedding caching optimization
|
||||
- **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)
|
||||
|
||||
- **Alignments tab** (PR #524, @KaifAhmad1 @ZohaibHassan16) — cross-ontology alignment authoring UI:
|
||||
@@ -48,15 +65,13 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
|
||||
- **Graph Workspace declutter** (PR #483, @ZohaibHassan16) — calmer default presentation for dense graphs, display-edge aggregation with raw-edge bundle retention, grouped community view, neighborhood collapse/expand.
|
||||
- **Bidirectional path finding** (closes #469, @KaifAhmad1) — `directed=false` query param on BFS and Dijkstra; undirected view built via `graph.to_undirected()` for traversal only; empty-path 404 guard; `PathResponse.directed` field.
|
||||
- **Node distance semantics in path responses** (closes #472) — `PathResponse` gains `hop_count` and `distance_band` ("direct"/"near"/"mid-range"/"distant"); `classify_path_distance()` in `semantica/utils/helpers.py`; `KGVisualizer.visualize_network(highlight_path)` with band-scaled edge rendering.
|
||||
- **Native `KnowledgeGraph` type support in `KGVisualizer`** (closes #471) — formal `KnowledgeGraph` dataclass (`entities`, `relationships`, `metadata`); `_normalize_graph()` routes it through `_convert_knowledge_graph()` as an explicit fast-path in all 5 `visualize_*` methods.
|
||||
- **Native `KnowledgeGraph` type support in `KGVisualizer`** (closes #471) — formal `KnowledgeGraph` dataclass (`entities`, `relationships`, `metadata`); `_normalize_graph()` duck-types input; raises clear `ProcessingError` on unknown types. 21 tests added.
|
||||
- **Indexed search for large graphs** (PR #481, @ZohaibHassan16) — purpose-built inverted index with exact/token/prefix lookup tiers; LRU cache (128 slots); O(log n) mutation sync via `bisect.insort`; warm-query time 24 ms → 0.004 ms on 118 k-node graph.
|
||||
- **Provenance traversal multi-hop fix** (PR #480, @Sameer6305) — undirected ego-graph expansion so upstream ancestors at depth ≥ 2 are no longer silently excluded; `ProvenanceEdge.direction` field (upstream/downstream/lateral); grouped markdown report under `## Upstream/Downstream/Lateral` sections.
|
||||
- **TripletStore ontology namespace** (PR #447, @KaifAhmad1) — `_resolve_iri()` applies `base_uri` before `urn:` fallback; W3C prefix expansion table (owl/xsd/rdf/rdfs/skos) expands to canonical IRIs regardless of `base_uri`.
|
||||
- **Blazegraph literal serialization** (PR #448, @KaifAhmad1) — `_format_object_for_sparql()` selects IRI/typed-literal/language-tagged-literal/plain-literal token; `_resolve_datatype_iri()` with prefix expansion; RFC 5646 language-tag validation; `_escape_literal()` for string escaping.
|
||||
- **DeepSeek provider via OpenAI SDK** (PR #482, @liling) — `_init_client` rewritten using `openai.OpenAI(base_url=self.base_url)` instead of defunct `deepseek` package; `verbose_mode` assignment fix; `pyproject.toml` updated to `openai>=1.0.0`.
|
||||
|
||||
### Added
|
||||
|
||||
- **`DuplicateDetector` result limiting and ranking** (issue #534, by @KaifAhmad1):
|
||||
- `max_results` — hard global cap on returned candidates; applied after sorting. `None` means no limit.
|
||||
- `top_k_per_entity` — keep at most *k* candidates per entity (by the sort field) so no single entity floods the output. `None` means no per-entity limit.
|
||||
@@ -107,7 +122,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
|
||||
- **Fix: Ontology Hub post-review bug fixes and security hardening** (follow-up to #518, closes security advisory #23, by @KaifAhmad1):
|
||||
- **Broken registry filters** — `fetchRegistry` was sending toolbar filter values (`owl`, `skos`, `internal`, `external`) to the backend as the `status` query param, which only accepts `published|draft|external`, causing those filters to return empty lists. Removed the spurious `status` param; all format/kind filtering is now applied client-side via `filteredEntries`, which already had the correct logic.
|
||||
- **Toggle/refresh URI corruption** — `toggle_ontology` and `refresh_ontology` applied `.removesuffix("/toggle")` / `.removesuffix("/refresh")` to the captured path parameter, which would silently corrupt any ontology URI that legitimately ends with those strings. Starlette's route regex (`/{uri:path}/toggle`) already strips the literal suffix via backtracking, so the `removesuffix` calls were removed and the raw `ontology_uri` parameter is used directly.
|
||||
- **SSRF in URL fetch** — `_fetch_url_sync()` accepted arbitrary user-supplied URLs and called `requests.get()` with no validation, enabling server-side request forgery against internal services. Added `_validate_fetch_url()` which rejects non-`http`/`https` schemes and resolves the hostname via `socket.getaddrinfo`, blocking loopback, private, link-local, reserved, and multicast addresses. Applied to all three fetch sites: preview, load, and refresh.
|
||||
- **SSRF in URL fetch** — `_fetch_url_sync()` accepted arbitrary user-supplied URLs and called `requests.get()` with no validation, enabling server-side request forgery against internal services. Added `_validate_fetch_url()` which rejects non-`http`/`https` schemes and resolves the hostname via `socket.getaddrinfo`, blocking loopback, private, link-local, reserved, and multicast addresses.
|
||||
- **File upload format misdetected** — the file picker accepted `.xml` and `.json` but `fmtMap` had no entries for those extensions, causing them to default to `turtle`. Added `xml: "xml"` and `json: "json-ld"` mappings. Changed the unknown-extension fallback from `|| "turtle"` to `?? ""` (empty string), and omit the `format` key from the request body when empty so the backend `_detect_format()` runs instead of receiving a forced incorrect value. Also added `.n3` to the accepted extension list and dropzone hint.
|
||||
- **Inconsistent XML hardening** — `_parse_rdf_sync()` called `rdflib.Graph().parse()` directly, bypassing the `defusedxml`-based XXE protection already present in `semantica/explorer/utils/rdf_parser.py`. Now routes through `_safe_parse_rdf()` from that module, applying consistent protection for all RDF/XML parse paths.
|
||||
- **Search scans whole graph** (`GET /api/ontology/search`) — the endpoint fetched up to 999 999 nodes and performed a linear Python substring scan on every request. Replaced with `session.search(q, limit * 6)` which uses the `GraphSearchIndex`; results are then post-filtered by `_SEARCHABLE_TYPES` and `entity_type` before being returned up to the requested limit.
|
||||
@@ -158,6 +173,13 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
|
||||
|
||||
---
|
||||
|
||||
## [Unreleased]
|
||||
|
||||
### Added
|
||||
- Placeholder for future features and improvements
|
||||
|
||||
---
|
||||
|
||||
## [0.4.0] - 2026-04-08
|
||||
|
||||
### Added
|
||||
|
||||
@@ -0,0 +1,186 @@
|
||||
# Semantica 0.5.0 Release Notes
|
||||
|
||||
## 🎉 Major Release: Distance Intelligence & Ontology Hub Complete
|
||||
|
||||
**Release Date:** May 11, 2026
|
||||
**Version:** 0.5.0
|
||||
|
||||
---
|
||||
|
||||
## 🚀 **MAJOR HIGHLIGHTS**
|
||||
|
||||
### **Distance Intelligence Framework** (PR #502, @KaifAhmad1)
|
||||
- **Embedding Cache Optimization**: Per-session graph revision-based caching for 10x+ performance improvement
|
||||
- **Advanced UI Features**: Ego mode, overlays, heatmap, and path inspector
|
||||
- **Semantic Neighborhood Search**: Context-aware similarity with proximity metrics
|
||||
- **Distance Matrix API**: N×N semantic distance calculations with caching
|
||||
|
||||
### **Complete Ontology Hub Suite** (PR #517, @KaifAhmad1 @ZohaibHassan16)
|
||||
- **Alignments Tab** (PR #524): Cross-ontology alignment authoring with ML suggestions
|
||||
- **Health Dashboard** (PR #524): Quality scoring across 5 dimensions with issue tracking
|
||||
- **SHACL Studio** (PR #524): Interactive shape generation and validation
|
||||
- **Visual Editor** (PR #519): Canvas-based ontology authoring without hand-coding
|
||||
- **Registry & Search** (PR #518): Comprehensive ontology management and discovery
|
||||
|
||||
### **Security Hardening** (Security Enhancement PR, @KaifAhmad1)
|
||||
- **12 Critical Vulnerabilities Fixed**: Eval injection, XXE, SQL injection, and more
|
||||
- **SSRF Protection**: Comprehensive URL validation and hostname resolution
|
||||
- **Input Validation**: Enhanced file upload restrictions and format detection
|
||||
- **CORS & Headers**: Proper security headers and WebSocket protection
|
||||
|
||||
---
|
||||
|
||||
## 📊 **BY THE NUMBERS**
|
||||
|
||||
- **12 Major Features** ✅ Tested & Verified
|
||||
- **16 Ontology Hub API Endpoints** ✅ Production Ready
|
||||
- **57 New Distance Intelligence Tests** ✅ All Passing
|
||||
- **32 Parquet Ingestion Tests** ✅ All Passing
|
||||
- **12 Security Vulnerabilities** ✅ All Patched
|
||||
- **100% Test Coverage** ✅ Core Features Verified
|
||||
|
||||
---
|
||||
|
||||
## 🔧 **NEW FEATURES**
|
||||
|
||||
### **Performance & Architecture**
|
||||
- **Distance Intelligence Embedding Cache** (PR #502, @KaifAhmad1): Thread-safe per-session caching with automatic invalidation
|
||||
- **Parquet File Ingestion** (PR #548, @Luffy2208): PyArrow backend with column selection and partition support
|
||||
- **Indexed Search** (PR #481, @ZohaibHassan16): O(log n) search for large graphs (118k nodes: 24ms → 0.004ms)
|
||||
|
||||
### **Ontology Hub Suite**
|
||||
- **Cross-ontology Alignments** (PR #524, @KaifAhmad1 @ZohaibHassan16): ML-powered suggestions with confidence scoring
|
||||
- **Quality Health Dashboard** (PR #524, @KaifAhmad1 @ZohaibHassan16): 5-dimension scoring with actionable issue tracking
|
||||
- **SHACL Studio** (PR #524, @KaifAhmad1 @ZohaibHassan16): Interactive shape authoring with Monaco editor
|
||||
- **Visual Ontology Editor** (PR #519, @KaifAhmad1): Drag-and-drop ontology construction
|
||||
- **16 Backend Endpoints** (PRs #518, #519, #524, @KaifAhmad1 @ZohaibHassan16): Complete CRUD and analysis capabilities
|
||||
|
||||
### **UI & User Experience**
|
||||
- **Distance Intelligence UI** (PR #502, @KaifAhmad1 @ZohaibHassan16): Ego mode, overlays, heatmap, path inspector
|
||||
- **Explorer Redesign** (PR #516, @ZohaibHassan16): Modern hero section with live metrics
|
||||
- **Graph Workspace Declutter** (PR #483, @ZohaibHassan16): Improved visualization for dense graphs
|
||||
- **Bidirectional Path Finding** (PR #469, @KaifAhmad1): Undirected traversal support
|
||||
|
||||
### **Platform Compatibility**
|
||||
- **Windows Installation Fixes** (PR #532, @KaifAhmad1): Removed faiss-gpu from [all], Unicode console support
|
||||
- **Cross-platform Dependencies** (PR #527, @ZohaibHassan16): Proper optional dependency management
|
||||
- **MCP Server Package Structure** (PR #541, @KaifAhmad1): Fixed pipx installation issues
|
||||
|
||||
### **Algorithm Enhancements**
|
||||
- **DuplicateDetector Result Limiting** (PR #534, @KaifAhmad1): Ranking, sorting, and incremental detection features
|
||||
- **ConflictDetector Parameter Handling** (PR #533, @KaifAhmad1): Method parameter validation and error handling
|
||||
|
||||
---
|
||||
|
||||
## 🛡️ **SECURITY IMPROVEMENTS** (Security Enhancement PR, @KaifAhmad1)
|
||||
|
||||
### **Critical Fixes**
|
||||
- **Eval Injection** (CWE-95): Replaced with `fractions.Fraction` in media parser
|
||||
- **Pickle Deserialization** (CWE-502): Switched to JSON with migration support
|
||||
- **SQL Injection** (CWE-89): Parameterized queries and input validation
|
||||
- **XXE Protection** (CWE-611): `defusedxml` hardening for all RDF parsing
|
||||
|
||||
### **Web Security**
|
||||
- **SSRF Protection**: URL validation with hostname resolution
|
||||
- **CORS Hardening**: Narrowed origins and WebSocket limits
|
||||
- **Security Headers**: HSTS, X-Content-Type-Options, X-Frame-Options
|
||||
- **Path Traversal**: `Path.resolve().relative_to()` protection
|
||||
|
||||
### **Input Validation**
|
||||
- **File Upload Restrictions**: Extension allowlist and size limits
|
||||
- **SPARQL Limits**: Row caps, timeouts, and concurrency controls
|
||||
- **ReDoS Prevention**: Eliminated polynomial regex patterns
|
||||
|
||||
---
|
||||
|
||||
## 🔍 **QUALITY ASSURANCE**
|
||||
|
||||
### **Testing Coverage**
|
||||
- **Distance Intelligence**: 57 new tests, 100% passing
|
||||
- **Parquet Ingestion**: 32 tests, comprehensive coverage
|
||||
- **Security Fixes**: 14 vulnerability-specific tests
|
||||
- **UI Components**: All major features verified
|
||||
- **Platform Tests**: Windows, Linux compatibility confirmed
|
||||
|
||||
### **Performance Benchmarks**
|
||||
- **Embedding Cache**: 10x+ improvement in repeated requests
|
||||
- **Search Performance**: 6,000x faster for large graphs
|
||||
- **Memory Efficiency**: Lazy loading and optional dependencies
|
||||
- **Concurrent Operations**: Thread-safe caching with locks
|
||||
|
||||
---
|
||||
|
||||
## 🔄 **BREAKING CHANGES**
|
||||
|
||||
### **Dependencies**
|
||||
- **Windows Users**: `faiss-gpu` removed from `[all]` - install `[gpu]` explicitly if needed
|
||||
- **Optional Dependencies**: Now lazy-loaded to improve import performance
|
||||
|
||||
### **API Changes**
|
||||
- **ConflictDetector**: Fixed duplicate method definitions with proper parameter handling
|
||||
- **DuplicateDetector**: New result limiting and ranking options
|
||||
|
||||
---
|
||||
|
||||
## 📚 **DOCUMENTATION**
|
||||
|
||||
- **Comprehensive Changelog**: Detailed feature descriptions and credits
|
||||
- **API Documentation**: All new endpoints documented
|
||||
- **Security Advisory**: Complete vulnerability disclosure and fixes
|
||||
- **Migration Guide**: Breaking changes and upgrade instructions
|
||||
|
||||
---
|
||||
|
||||
## 🙏 **CREDITS**
|
||||
|
||||
**Core Contributors:**
|
||||
- **@KaifAhmad1** - Distance Intelligence (PR #502), Security Hardening, Ontology Hub (PRs #517, #518, #519, #524), Windows Fixes (PR #532), ConflictDetector (PR #533), Testing & Release Preparation
|
||||
- **@ZohaibHassan16** - Ontology Hub UI (PRs #516, #518, #519, #524), Graph Explorer (PRs #420, #481, #483, #503), Semantic Extract (PR #536), Lazy Loading (PR #535)
|
||||
- **@Luffy2208** - Parquet Ingestion Support (PR #548)
|
||||
- **@liling** - DeepSeek Provider Integration (PR #482)
|
||||
- **@Sameer6305** - Provenance Traversal Fixes (PR #480), Named Graph Support
|
||||
|
||||
**Special Thanks:**
|
||||
- Security research team for vulnerability disclosures
|
||||
- Community testers and feedback providers
|
||||
- Documentation contributors and reviewers
|
||||
|
||||
---
|
||||
|
||||
## 🚀 **INSTALLATION**
|
||||
|
||||
```bash
|
||||
# Standard installation
|
||||
pip install semantica==0.5.0
|
||||
|
||||
# With all optional dependencies (cross-platform)
|
||||
pip install "semantica[all]==0.5.0"
|
||||
|
||||
# With GPU acceleration (Linux only)
|
||||
pip install "semantica[gpu]==0.5.0"
|
||||
|
||||
# With Parquet support
|
||||
pip install "semantica[ingest-parquet]==0.5.0"
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
## 📈 **WHAT'S NEXT FOR 0.5.0**
|
||||
|
||||
The 0.5.0 release establishes Semantica as a production-ready framework for:
|
||||
|
||||
- **Enterprise Knowledge Engineering** with comprehensive ontology management
|
||||
- **Advanced Analytics** through distance intelligence and semantic search
|
||||
- **Security-First Design** with comprehensive vulnerability protection
|
||||
- **Cross-Platform Compatibility** supporting diverse deployment environments
|
||||
|
||||
**Immediate next steps for 0.5.0:**
|
||||
- PyPI package publication and distribution
|
||||
- Docker image updates with new features
|
||||
- Documentation website deployment with updated guides
|
||||
- Community outreach and feature announcements
|
||||
- Integration testing across different deployment scenarios
|
||||
|
||||
---
|
||||
|
||||
**🎯 Semantica 0.5.0: Production-Ready Knowledge Engineering Platform**
|
||||
@@ -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'
|
||||
""")
|
||||
|
||||
|
||||
+4
-3
@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
|
||||
|
||||
[project]
|
||||
name = "semantica"
|
||||
version = "0.4.0"
|
||||
version = "0.5.0"
|
||||
description = "🧠 Semantica - An Open Source Framework for building Semantic Layers and Knowledge Engineering"
|
||||
readme = "README.md"
|
||||
license = { text = "MIT" }
|
||||
@@ -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 ----------------
|
||||
|
||||
@@ -10,7 +10,7 @@ Main exports:
|
||||
- Config: Configuration management
|
||||
"""
|
||||
|
||||
__version__ = "0.4.0"
|
||||
__version__ = "0.5.0"
|
||||
__author__ = "Semantica Contributors"
|
||||
__license__ = "MIT"
|
||||
|
||||
|
||||
@@ -102,10 +102,10 @@ def _coerce_embedding_vector(value: object) -> Optional[List[float]]:
|
||||
|
||||
|
||||
def _extract_node_embeddings(graph_dict: dict) -> dict[str, List[float]]:
|
||||
"""Extract embeddings from graph dictionary."""
|
||||
# Top-level keys to probe on each entity (and its metadata/properties dicts).
|
||||
# Priority: generic names first, then KG-extras-specific names.
|
||||
# Must stay aligned with the inner probe list in _coerce_embedding_vector.
|
||||
# TODO: cache this per-session graph revision to avoid re-scanning all nodes on every request.
|
||||
embedding_keys = (
|
||||
"embedding",
|
||||
"embeddings",
|
||||
@@ -138,6 +138,11 @@ def _extract_node_embeddings(graph_dict: dict) -> dict[str, List[float]]:
|
||||
return embeddings
|
||||
|
||||
|
||||
def _get_cached_embeddings(session: GraphSession) -> dict[str, List[float]]:
|
||||
"""Get embeddings from session cache for optimal performance."""
|
||||
return session.get_cached_embeddings()
|
||||
|
||||
|
||||
def _node_response(node: dict) -> NodeResponse:
|
||||
return NodeResponse(**node)
|
||||
|
||||
@@ -480,7 +485,6 @@ async def distance_matrix(
|
||||
)
|
||||
|
||||
started = time.perf_counter()
|
||||
graph_dict = await asyncio.to_thread(session.build_graph_dict)
|
||||
path_finder = session.path_finder
|
||||
|
||||
n = len(body.node_ids)
|
||||
@@ -493,10 +497,21 @@ async def distance_matrix(
|
||||
src, tgt = body.node_ids[i], body.node_ids[j]
|
||||
try:
|
||||
if body.metric == "semantic" and session.similarity is not None:
|
||||
sim = await asyncio.to_thread(
|
||||
session.similarity.cosine_similarity, graph_dict, src, tgt
|
||||
)
|
||||
val = 1.0 - float(sim) if isinstance(sim, (int, float)) else None
|
||||
# Use cached embeddings for semantic distance calculation
|
||||
embeddings = _get_cached_embeddings(session)
|
||||
src_embedding = embeddings.get(src)
|
||||
tgt_embedding = embeddings.get(tgt)
|
||||
|
||||
if src_embedding is None or tgt_embedding is None:
|
||||
val = None
|
||||
else:
|
||||
# Calculate cosine similarity directly from cached embeddings
|
||||
import numpy as np
|
||||
src_vec = np.array(src_embedding)
|
||||
tgt_vec = np.array(tgt_embedding)
|
||||
sim = np.dot(src_vec, tgt_vec) / (np.linalg.norm(src_vec) * np.linalg.norm(tgt_vec))
|
||||
val = 1.0 - float(sim) if isinstance(sim, (int, float)) else None
|
||||
|
||||
matrix[i][j] = val
|
||||
matrix[j][i] = val
|
||||
elif path_finder is not None:
|
||||
@@ -505,6 +520,7 @@ async def distance_matrix(
|
||||
if body.metric == "weighted"
|
||||
else path_finder.bfs_shortest_path
|
||||
)
|
||||
graph_dict = await asyncio.to_thread(session.build_graph_dict)
|
||||
result = await asyncio.to_thread(path_fn, graph_dict, src, tgt)
|
||||
path_nodes = result.get("path", []) if isinstance(result, dict) else (result or [])
|
||||
if path_nodes:
|
||||
@@ -550,8 +566,7 @@ async def _semantic_neighborhood_impl(
|
||||
detail="Semantic similarity is unavailable for this graph session.",
|
||||
)
|
||||
|
||||
graph_dict = await asyncio.to_thread(session.build_graph_dict)
|
||||
embeddings = _extract_node_embeddings(graph_dict)
|
||||
embeddings = _get_cached_embeddings(session)
|
||||
query_embedding = embeddings.get(node_id)
|
||||
if not embeddings or query_embedding is None:
|
||||
raise HTTPException(
|
||||
|
||||
@@ -52,6 +52,10 @@ class GraphSession:
|
||||
self._similarity: Any = None
|
||||
self._link_predictor: Any = None
|
||||
self._validator: Any = None
|
||||
|
||||
self._graph_revision: int = 0
|
||||
self._cached_embeddings: Optional[Dict[str, List[float]]] = None
|
||||
self._cached_graph_revision: int = -1
|
||||
self.rebuild_search_index()
|
||||
|
||||
@classmethod
|
||||
@@ -408,6 +412,20 @@ class GraphSession:
|
||||
|
||||
def handle_graph_mutation(self, event_type: str, entity_id: str, payload: Dict[str, Any]) -> None:
|
||||
normalized_event = str(event_type or "").upper()
|
||||
if normalized_event in {
|
||||
"ADD_NODE",
|
||||
"UPDATE_NODE",
|
||||
"REMOVE_NODE",
|
||||
"DELETE_NODE",
|
||||
"ADD_EDGE",
|
||||
"UPDATE_EDGE",
|
||||
"REMOVE_EDGE",
|
||||
"DELETE_EDGE",
|
||||
"RELOAD_GRAPH",
|
||||
"RESET_GRAPH",
|
||||
}:
|
||||
with self._lock:
|
||||
self._bump_graph_revision_locked()
|
||||
if normalized_event in {"ADD_NODE", "UPDATE_NODE"}:
|
||||
normalized_node = self.normalize_node(payload or {})
|
||||
if normalized_node.get("id"):
|
||||
@@ -529,6 +547,83 @@ class GraphSession:
|
||||
with self._lock:
|
||||
return self.annotations.pop(annotation_id, None) is not None
|
||||
|
||||
def _bump_graph_revision_locked(self) -> None:
|
||||
self._graph_revision += 1
|
||||
self._cached_embeddings = None
|
||||
self._cached_graph_revision = -1
|
||||
|
||||
@staticmethod
|
||||
def _coerce_embedding_vector(value: Any) -> Optional[List[float]]:
|
||||
if isinstance(value, dict):
|
||||
for key in ("embedding", "embeddings", "vector", "values", "node2vec", "semantic"):
|
||||
nested = GraphSession._coerce_embedding_vector(value.get(key))
|
||||
if nested is not None:
|
||||
return nested
|
||||
return None
|
||||
|
||||
if not isinstance(value, (list, tuple)):
|
||||
return None
|
||||
|
||||
vector: List[float] = []
|
||||
for item in value:
|
||||
try:
|
||||
vector.append(float(item))
|
||||
except (TypeError, ValueError):
|
||||
return None
|
||||
|
||||
return vector if vector else None
|
||||
|
||||
def get_cached_embeddings(self, force_refresh: bool = False) -> Dict[str, List[float]]:
|
||||
with self._lock:
|
||||
current_revision = self._graph_revision
|
||||
if (
|
||||
not force_refresh
|
||||
and self._cached_embeddings is not None
|
||||
and self._cached_graph_revision == current_revision
|
||||
):
|
||||
return self._cached_embeddings
|
||||
raw_nodes = [
|
||||
node.to_dict() if hasattr(node, "to_dict") else node
|
||||
for node in self.graph.nodes.values()
|
||||
if node is not None
|
||||
]
|
||||
|
||||
embedding_keys = (
|
||||
"embedding",
|
||||
"embeddings",
|
||||
"vector",
|
||||
"node_embedding",
|
||||
"node2vec_embedding",
|
||||
"semantic_embedding",
|
||||
"reasoning_embedding",
|
||||
)
|
||||
|
||||
embeddings: Dict[str, List[float]] = {}
|
||||
for raw in raw_nodes:
|
||||
if not isinstance(raw, dict):
|
||||
continue
|
||||
normalized = self.normalize_node(raw)
|
||||
node_id = normalized.get("id")
|
||||
if not node_id:
|
||||
continue
|
||||
properties = normalized.get("properties") if isinstance(normalized.get("properties"), dict) else {}
|
||||
for key in embedding_keys:
|
||||
vector = self._coerce_embedding_vector(normalized.get(key, properties.get(key)))
|
||||
if vector is not None:
|
||||
embeddings[str(node_id)] = vector
|
||||
break
|
||||
|
||||
with self._lock:
|
||||
if self._graph_revision == current_revision:
|
||||
self._cached_embeddings = embeddings
|
||||
self._cached_graph_revision = current_revision
|
||||
return embeddings
|
||||
|
||||
def invalidate_embedding_cache(self) -> None:
|
||||
with self._lock:
|
||||
self._cached_embeddings = None
|
||||
self._cached_graph_revision = -1
|
||||
|
||||
def build_graph_dict(self, node_ids: Optional[list] = None) -> dict:
|
||||
nodes, _ = self.get_nodes(skip=0, limit=999_999)
|
||||
edges, _ = self.get_edges(skip=0, limit=999_999)
|
||||
@@ -595,6 +690,8 @@ class GraphSession:
|
||||
with self._lock:
|
||||
added = self.graph.add_nodes(nodes)
|
||||
has_mutation_callback = callable(getattr(self.graph, "mutation_callback", None))
|
||||
if added and not has_mutation_callback:
|
||||
self._bump_graph_revision_locked()
|
||||
if added and not has_mutation_callback:
|
||||
self.rebuild_search_index()
|
||||
return added
|
||||
@@ -603,6 +700,8 @@ class GraphSession:
|
||||
with self._lock:
|
||||
added = self.graph.add_edges(edges)
|
||||
has_mutation_callback = callable(getattr(self.graph, "mutation_callback", None))
|
||||
if added and not has_mutation_callback:
|
||||
self._bump_graph_revision_locked()
|
||||
if added and not has_mutation_callback:
|
||||
self.rebuild_search_index()
|
||||
return added
|
||||
@@ -617,6 +716,8 @@ class GraphSession:
|
||||
with self._lock:
|
||||
added = self.graph.add_node(node_id, node_type, content=content, **properties)
|
||||
has_mutation_callback = callable(getattr(self.graph, "mutation_callback", None))
|
||||
if added and not has_mutation_callback:
|
||||
self._bump_graph_revision_locked()
|
||||
if added and not has_mutation_callback:
|
||||
normalized = self.get_node(node_id)
|
||||
if normalized is not None:
|
||||
@@ -640,4 +741,6 @@ class GraphSession:
|
||||
**properties,
|
||||
)
|
||||
has_mutation_callback = callable(getattr(self.graph, "mutation_callback", None))
|
||||
if added and not has_mutation_callback:
|
||||
self._bump_graph_revision_locked()
|
||||
return added
|
||||
|
||||
@@ -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