# Story 1.2: DAG Project Structure & Configuration

**Epic:** Airflow Infrastructure & Parallel Deployment
**Status:** Done
**Priority:** High
**Estimate:** 2 hours

---

## User Story

As a **developer**, I want to **set up the project structure for Airflow DAGs and plugins** so that **we have a clean, organized codebase for the migration**.

---

## Acceptance Criteria

- [x] `dags/` directory created with proper structure
- [x] `plugins/operators/` directory created for custom operators
- [x] `plugins/hooks/` directory created for connection hooks
- [x] `plugins/models/` directory created for XCom data schemas
- [x] DAG config file created with shared settings
- [x] All directories are Python packages with `__init__.py`
- [x] Airflow can discover and load DAGs from the structure

---

## Technical Details

### Project Structure to Create

```
textiq-doc-extraction/
├── dags/
│   ├── __init__.py
│   ├── document_extraction_dag.py        # Main DAG (Story 1.6)
│   ├── large_table_extraction_dag.py     # Large table handling (future)
│   └── config.py                         # DAG configuration
├── plugins/
│   ├── __init__.py
│   ├── operators/
│   │   ├── __init__.py
│   │   ├── validate_event.py             # Story 1.3
│   │   ├── fetch_document.py             # Story 1.3
│   │   ├── parse_document.py             # Story 1.3
│   │   ├── detect_headers.py             # Story 1.4
│   │   ├── chunk_content.py              # Story 1.4
│   │   ├── generate_embeddings.py        # Story 1.4
│   │   ├── store_chunks.py               # Story 1.5
│   │   └── store_vectors.py              # Story 1.5
│   ├── hooks/
│   │   ├── __init__.py
│   │   ├── seaweedfs_hook.py             # SeaweedFS connection
│   │   └── milvus_hook.py                # Milvus connection
│   └── models/
│       ├── __init__.py
│       └── xcom_schemas.py               # XCom data structures
```

### Files to Create

| File | Description |
|------|-------------|
| `dags/__init__.py` | Package init |
| `dags/config.py` | Shared DAG configuration (default_args, etc.) |
| `plugins/__init__.py` | Package init |
| `plugins/operators/__init__.py` | Operators package with exports |
| `plugins/hooks/__init__.py` | Hooks package |
| `plugins/models/__init__.py` | Models package |
| `plugins/models/xcom_schemas.py` | XCom data structures |

---

## Tasks

- [x] Create `dags/` directory structure
- [x] Create `plugins/operators/` directory structure
- [x] Create `plugins/hooks/` directory structure
- [x] Create `plugins/models/` directory structure
- [x] Create `dags/config.py` with default_args
- [x] Create `plugins/models/xcom_schemas.py` with dataclasses
- [x] Create all `__init__.py` files
- [x] Verify Airflow can load plugins

---

## Implementation

### dags/config.py

```python
from datetime import timedelta

default_args = {
    'owner': 'textiq',
    'depends_on_past': False,
    'email_on_failure': True,
    'email_on_retry': False,
    'retries': 3,
    'retry_delay': timedelta(minutes=2),
    'retry_exponential_backoff': True,
    'max_retry_delay': timedelta(minutes=30),
}

# Task-specific timeouts
PARSE_TIMEOUT = timedelta(minutes=30)
EMBEDDING_TIMEOUT = timedelta(minutes=60)
STORE_TIMEOUT = timedelta(minutes=15)

# XCom settings
MAX_XCOM_SIZE = 48 * 1024  # 48KB, store larger in SeaweedFS
```

### plugins/models/xcom_schemas.py

```python
from dataclasses import dataclass
from typing import Any, Optional
from uuid import UUID

@dataclass
class ValidatedEvent:
    """Output from ValidateEventOperator."""
    document_id: UUID
    file_path: str
    content_type: str
    size_bytes: int
    event_id: str
    bucket: str
    object_key: str

@dataclass
class FetchedDocument:
    """Output from FetchDocumentOperator."""
    local_path: str
    content_type: str
    original_filename: str
    file_extension: str

@dataclass
class ParsedDocument:
    """Output from ParseDocumentOperator."""
    doc_type: str
    html_content: Optional[str]
    markdown_content: Optional[str]
    tables: list[dict]
    local_path: str

@dataclass
class ContentChunks:
    """Output from ChunkContentOperator."""
    chunks: list[dict]
    chunk_count: int
    chunking_strategy: str

@dataclass
class ChunkEmbeddings:
    """Output from GenerateEmbeddingsOperator."""
    embeddings: list[dict]
    embedding_count: int
    model: str
    cached_count: int
```

---

## Testing

```bash
# Verify directory structure
ls -la dags/ plugins/

# Test Python imports
python -c "from plugins.models.xcom_schemas import ValidatedEvent; print('OK')"

# Test Airflow plugin discovery
airflow plugins
```

---

## Notes

- All directories must be Python packages for Airflow to discover them
- XCom schemas use dataclasses for type safety
- Config file centralizes settings for easy modification

---

## File List

| File | Action | Description |
|------|--------|-------------|
| `airflow/dags/__init__.py` | Created | DAGs package init |
| `airflow/dags/config.py` | Created | Shared DAG configuration with default_args and timeouts |
| `airflow/plugins/__init__.py` | Created | TextIQ Airflow plugin package (conditional Airflow import) |
| `airflow/plugins/operators/__init__.py` | Created | Custom operators package init |
| `airflow/plugins/hooks/__init__.py` | Created | Custom hooks package init |
| `airflow/plugins/models/__init__.py` | Created | Data models package init with exports |
| `airflow/plugins/models/xcom_schemas.py` | Created | XCom dataclasses (ValidatedEvent, FetchedDocument, ParsedDocument, ContentChunks, ChunkEmbeddings) |

---

## Dev Agent Record

### Implementation Notes

- Created complete Airflow project structure in `airflow/` directory
- All directories are proper Python packages with `__init__.py` files
- DAG config includes default_args with retry logic, timeouts for different task types, and XCom size limits
- XCom schemas implemented as dataclasses for type safety
- Plugin package uses conditional import for Airflow to allow local testing without Airflow installed
- Directory structure follows Airflow best practices with separation of concerns (operators, hooks, models)

### Validation Results

All tests passed:
- ✓ Directory structure created correctly
- ✓ Python imports work locally (with uv)
- ✓ DAG config imports successfully in Airflow container
- ✓ XCom schemas import successfully in Airflow container
- ✓ Plugin package loads without errors
- ✓ Airflow scheduler can discover DAGs directory (no DAGs yet, as expected)

---

## Change Log

| Date | Change |
|------|--------|
| 2025-12-09 | Initial implementation of Airflow DAG project structure |
| 2025-12-09 | Verified all acceptance criteria - story complete |