Data Classification and Storage Mapping:

┌─────────────────────────────────────────────────────┐
│                coomia-dip Data Layer                   │
│                                                       │
│  ┌──────────────────────┐  ┌────────────────────────┐ │
│  │  Structured Data      │  │  Unstructured Data      │ │
│  │                        │  │                          │ │
│  │  • Ontology entities   │  │  • Pipeline output files │ │
│  │  • Relationship edges  │  │  • ML model weights      │ │
│  │  • Event streams       │  │  • Export reports (CSV)   │ │
│  │  • Metric time-series  │  │  • User uploads          │ │
│  │  • Audit logs          │  │  • Visualization images  │ │
│  │                        │  │  • Large JSON/Parquet    │ │
│  │  Store: Doris+Iceberg  │  │  Store: MinIO (S3-compat)│ │
│  └──────────────────────┘  └────────────────────────┘ │
└─────────────────────────────────────────────────────┘

MinIO Cluster Topology (coomia-dip Production):

┌──────────────────────────────────────────────────────────┐
│                     Nginx / Traefik                       │
│              (TLS Termination + Load Balancing)            │
│         :9000 (API)        :9001 (Console)                │
└─────────┬──────────────────────┬─────────────────────────┘
          │                      │
    ┌─────┴─────┐          ┌─────┴─────┐
    │           │          │           │
┌───┴───┐ ┌───┴───┐ ┌───┴───┐ ┌───┴───┐
│MinIO-1│ │MinIO-2│ │MinIO-3│ │MinIO-4│
│ Node  │ │ Node  │ │ Node  │ │ Node  │
│       │ │       │ │       │ │       │
│/data1 │ │/data1 │ │/data1 │ │/data1 │
│/data2 │ │/data2 │ │/data2 │ │/data2 │
└───────┘ └───────┘ └───────┘ └───────┘

Erasure Coding Group: EC(4,2) — 4 data + 2 parity
Tolerance: Any 2 node failures without data loss

# deployment-Layer/docker-compose-minio.yml
version: '3.8'

services:
  minio-1:
    image: quay.io/minio/minio:RELEASE.2024-06-13T22-53-53Z
    command: server --console-address ":9001" http://minio-{1...4}/data{1...2}
    environment:
      MINIO_ROOT_USER: ${MINIO_ROOT_USER}
      MINIO_ROOT_PASSWORD: ${MINIO_ROOT_PASSWORD}
      MINIO_PROMETHEUS_AUTH_TYPE: public
      MINIO_SCANNER_SPEED: slow
    volumes:
      - minio1-data1:/data1
      - minio1-data2:/data2
    networks:
      - coomia-dip-net
    healthcheck:
      test: ["CMD", "mc", "ready", "local"]
      interval: 30s
      timeout: 10s
      retries: 3
    deploy:
      resources:
        limits:
          memory: 4G
        reservations:
          memory: 2G

  minio-2:
    image: quay.io/minio/minio:RELEASE.2024-06-13T22-53-53Z
    command: server --console-address ":9001" http://minio-{1...4}/data{1...2}
    environment:
      MINIO_ROOT_USER: ${MINIO_ROOT_USER}
      MINIO_ROOT_PASSWORD: ${MINIO_ROOT_PASSWORD}
    volumes:
      - minio2-data1:/data1
      - minio2-data2:/data2
    networks:
      - coomia-dip-net

  minio-3:
    image: quay.io/minio/minio:RELEASE.2024-06-13T22-53-53Z
    command: server --console-address ":9001" http://minio-{1...4}/data{1...2}
    environment:
      MINIO_ROOT_USER: ${MINIO_ROOT_USER}
      MINIO_ROOT_PASSWORD: ${MINIO_ROOT_PASSWORD}
    volumes:
      - minio3-data1:/data1
      - minio3-data2:/data2
    networks:
      - coomia-dip-net

  minio-4:
    image: quay.io/minio/minio:RELEASE.2024-06-13T22-53-53Z
    command: server --console-address ":9001" http://minio-{1...4}/data{1...2}
    environment:
      MINIO_ROOT_USER: ${MINIO_ROOT_USER}
      MINIO_ROOT_PASSWORD: ${MINIO_ROOT_PASSWORD}
    volumes:
      - minio4-data1:/data1
      - minio4-data2:/data2
    networks:
      - coomia-dip-net

volumes:
  minio1-data1:
  minio1-data2:
  minio2-data1:
  minio2-data2:
  minio3-data1:
  minio3-data2:
  minio4-data1:
  minio4-data2:

networks:
  coomia-dip-net:
    external: true

# deployment-Layer/docker-compose-minio-dev.yml
version: '3.8'

services:
  minio:
    image: quay.io/minio/minio:RELEASE.2024-06-13T22-53-53Z
    command: server /data --console-address ":9001"
    ports:
      - "9000:9000"
      - "9001:9001"
    environment:
      MINIO_ROOT_USER: coomia-dip
      MINIO_ROOT_PASSWORD: coomia-dip-dev-2024
    volumes:
      - minio-data:/data
    networks:
      - coomia-dip-net

volumes:
  minio-data:

Bucket Naming Convention:

onto-{project_id}-{category}

Examples:
  onto-proj001-pipeline    # Pipeline outputs
  onto-proj001-artifacts   # ML model artifacts
  onto-proj001-exports     # Export files
  onto-proj001-uploads     # User uploads
  onto-system-shared       # System shared resources
  onto-system-templates    # Template files

Internal Bucket Directory Structure:

onto-proj001-pipeline/
├── runs/
│   ├── run-20240615-001/
│   │   ├── output/
│   │   │   ├── result.parquet
│   │   │   └── summary.json
│   │   ├── logs/
│   │   │   ├── stdout.log
│   │   │   └── stderr.log
│   │   └── _metadata.json
│   └── run-20240615-002/
│       └── ...
└── snapshots/
    └── 2024-06-15/
        └── full-export.parquet

onto-proj001-artifacts/
├── models/
│   ├── risk-scorer-v1.0/
│   │   ├── model.onnx
│   │   ├── config.json
│   │   ├── tokenizer.json
│   │   └── _manifest.json
│   └── anomaly-detector-v2.1/
│       └── ...
├── checkpoints/
│   └── training-run-001/
│       ├── epoch-10.pt
│       └── epoch-20.pt
└── evaluations/
    └── risk-scorer-v1.0/
        ├── metrics.json
        └── confusion-matrix.png

# python-sdk/ontology_sdk/storage/minio_policy.py
from dataclasses import dataclass
from typing import Any


@dataclass
class BucketPolicy:
    """MinIO bucket access policy generator"""

    project_id: str

    def generate_readwrite_policy(self) -> dict[str, Any]:
        """Generate project read-write policy"""
        return {
            "Version": "2012-10-17",
            "Statement": [
                {
                    "Effect": "Allow",
                    "Action": [
                        "s3:GetObject",
                        "s3:PutObject",
                        "s3:DeleteObject",
                        "s3:ListBucket",
                    ],
                    "Resource": [
                        f"arn:aws:s3:::onto-{self.project_id}-*",
                        f"arn:aws:s3:::onto-{self.project_id}-*/*",
                    ],
                },
                {
                    "Effect": "Deny",
                    "Action": ["s3:*"],
                    "Resource": ["arn:aws:s3:::onto-system-*"],
                },
            ],
        }

    def generate_readonly_policy(self) -> dict[str, Any]:
        """Generate project read-only policy"""
        return {
            "Version": "2012-10-17",
            "Statement": [
                {
                    "Effect": "Allow",
                    "Action": [
                        "s3:GetObject",
                        "s3:ListBucket",
                    ],
                    "Resource": [
                        f"arn:aws:s3:::onto-{self.project_id}-*",
                        f"arn:aws:s3:::onto-{self.project_id}-*/*",
                    ],
                }
            ],
        }

    def generate_pipeline_policy(self) -> dict[str, Any]:
        """Generate pipeline-specific policy (write only to pipeline bucket)"""
        return {
            "Version": "2012-10-17",
            "Statement": [
                {
                    "Effect": "Allow",
                    "Action": [
                        "s3:GetObject",
                        "s3:PutObject",
                        "s3:ListBucket",
                    ],
                    "Resource": [
                        f"arn:aws:s3:::onto-{self.project_id}-pipeline",
                        f"arn:aws:s3:::onto-{self.project_id}-pipeline/*",
                    ],
                },
                {
                    "Effect": "Allow",
                    "Action": ["s3:GetObject"],
                    "Resource": [
                        f"arn:aws:s3:::onto-{self.project_id}-artifacts/*",
                    ],
                },
            ],
        }

Pipeline Output Flow:

┌──────────┐    ┌──────────┐    ┌──────────────┐    ┌─────────┐
│ Pipeline │    │ Compute  │    │   MinIO      │    │  Doris  │
│ Engine   │───>│ Worker   │───>│   Storage    │───>│ Catalog │
│(Schedule)│    │(Executor)│    │ (Large Files)│    │(Metadata│
└──────────┘    └────┬─────┘    └──────────────┘    └─────────┘
                     │
                     │ Simultaneous write
                     ▼
              ┌──────────────┐
              │ Iceberg Table│
              │(Structured   │
              │ Results)     │
              └──────────────┘

# intelligence-Layer/pipeline/output_writer.py
import json
import hashlib
from datetime import datetime, timezone
from pathlib import PurePosixPath
from typing import BinaryIO

from minio import Minio
from pydantic import BaseModel, Field


class PipelineOutputMeta(BaseModel):
    """Pipeline output metadata"""
    run_id: str
    pipeline_id: str
    project_id: str
    output_type: str  # "parquet", "csv", "json", "binary"
    file_name: str
    file_size: int
    md5_hash: str
    created_at: datetime = Field(default_factory=lambda: datetime.now(timezone.utc))
    tags: dict[str, str] = Field(default_factory=dict)


class PipelineOutputWriter:
    """Writer for pipeline outputs to MinIO"""

    def __init__(self, minio_client: Minio):
        self._client = minio_client

    def write_output(
        self,
        project_id: str,
        run_id: str,
        pipeline_id: str,
        file_name: str,
        data: BinaryIO,
        content_type: str = "application/octet-stream",
        tags: dict[str, str] | None = None,
    ) -> PipelineOutputMeta:
        """Write pipeline output file to MinIO"""
        bucket_name = f"onto-{project_id}-pipeline"
        self._ensure_bucket(bucket_name)

        # Build object path
        object_path = str(
            PurePosixPath("runs") / run_id / "output" / file_name
        )

        # Read data and compute hash
        content = data.read()
        md5_hash = hashlib.md5(content).hexdigest()

        # Upload to MinIO
        from io import BytesIO

        self._client.put_object(
            bucket_name=bucket_name,
            object_name=object_path,
            data=BytesIO(content),
            length=len(content),
            content_type=content_type,
            metadata={
                "x-amz-meta-run-id": run_id,
                "x-amz-meta-pipeline-id": pipeline_id,
                "x-amz-meta-md5": md5_hash,
            },
        )

        # Build metadata
        meta = PipelineOutputMeta(
            run_id=run_id,
            pipeline_id=pipeline_id,
            project_id=project_id,
            output_type=file_name.rsplit(".", 1)[-1] if "." in file_name else "binary",
            file_name=file_name,
            file_size=len(content),
            md5_hash=md5_hash,
            tags=tags or {},
        )

        # Write metadata file
        meta_path = str(
            PurePosixPath("runs") / run_id / "_metadata.json"
        )
        meta_bytes = meta.model_dump_json(indent=2).encode("utf-8")
        self._client.put_object(
            bucket_name=bucket_name,
            object_name=meta_path,
            data=BytesIO(meta_bytes),
            length=len(meta_bytes),
            content_type="application/json",
        )

        return meta

    def write_log(
        self,
        project_id: str,
        run_id: str,
        log_type: str,  # "stdout" or "stderr"
        content: str,
    ) -> None:
        """Write pipeline execution log"""
        bucket_name = f"onto-{project_id}-pipeline"
        self._ensure_bucket(bucket_name)

        object_path = str(
            PurePosixPath("runs") / run_id / "logs" / f"{log_type}.log"
        )
        data = content.encode("utf-8")
        from io import BytesIO

        self._client.put_object(
            bucket_name=bucket_name,
            object_name=object_path,
            data=BytesIO(data),
            length=len(data),
            content_type="text/plain",
        )

    def _ensure_bucket(self, bucket_name: str) -> None:
        """Ensure bucket exists"""
        if not self._client.bucket_exists(bucket_name):
            self._client.make_bucket(bucket_name)

ML Model Artifact Lifecycle:

  Training Phase        Registration Phase     Deployment Phase       Archive Phase
┌──────────┐       ┌──────────┐       ┌──────────┐       ┌──────────┐
│ Training │──────>│ Registry │──────>│ Serving  │──────>│ Archive  │
│          │       │          │       │          │       │          │
│ Checkpts │       │ Version  │       │ Active   │       │ Cold     │
│ in MinIO │       │ Tag      │       │ Download │       │ Storage  │
└──────────┘       └──────────┘       └──────────┘       └──────────┘
     │                   │                  │                  │
     ▼                   ▼                  ▼                  ▼
  artifacts/          artifacts/         Presigned          Lifecycle
  checkpoints/        models/            URL Download       Rule Transition
  train-run-*/        model-v*/          (Time-limited)     (After 90 days)

# intelligence-Layer/ml/model_registry.py
import json
from datetime import datetime, timezone, timedelta
from enum import Enum
from io import BytesIO
from typing import BinaryIO

from minio import Minio
from pydantic import BaseModel, Field


class ModelStage(str, Enum):
    """Model lifecycle stage"""
    DEVELOPMENT = "development"
    STAGING = "staging"
    PRODUCTION = "production"
    ARCHIVED = "archived"


class ModelManifest(BaseModel):
    """Model manifest"""
    model_name: str
    version: str
    stage: ModelStage = ModelStage.DEVELOPMENT
    framework: str  # "pytorch", "onnx", "sklearn", "xgboost"
    description: str = ""
    metrics: dict[str, float] = Field(default_factory=dict)
    parameters: dict[str, str] = Field(default_factory=dict)
    files: list[str] = Field(default_factory=list)
    created_at: datetime = Field(default_factory=lambda: datetime.now(timezone.utc))
    created_by: str = ""
    tags: dict[str, str] = Field(default_factory=dict)


class ModelRegistry:
    """MinIO-backed model registry"""

    def __init__(self, minio_client: Minio):
        self._client = minio_client

    def register_model(
        self,
        project_id: str,
        model_name: str,
        version: str,
        framework: str,
        files: dict[str, BinaryIO],
        metrics: dict[str, float] | None = None,
        parameters: dict[str, str] | None = None,
        description: str = "",
        created_by: str = "",
    ) -> ModelManifest:
        """Register a new model version"""
        bucket_name = f"onto-{project_id}-artifacts"
        if not self._client.bucket_exists(bucket_name):
            self._client.make_bucket(bucket_name)

        # Upload all model files
        file_list: list[str] = []
        for filename, fileobj in files.items():
            object_path = f"models/{model_name}-{version}/{filename}"
            content = fileobj.read()
            self._client.put_object(
                bucket_name=bucket_name,
                object_name=object_path,
                data=BytesIO(content),
                length=len(content),
            )
            file_list.append(filename)

        # Create and upload manifest
        manifest = ModelManifest(
            model_name=model_name,
            version=version,
            framework=framework,
            description=description,
            metrics=metrics or {},
            parameters=parameters or {},
            files=file_list,
            created_by=created_by,
        )

        manifest_path = f"models/{model_name}-{version}/_manifest.json"
        manifest_bytes = manifest.model_dump_json(indent=2).encode("utf-8")
        self._client.put_object(
            bucket_name=bucket_name,
            object_name=manifest_path,
            data=BytesIO(manifest_bytes),
            length=len(manifest_bytes),
            content_type="application/json",
        )

        return manifest

    def promote_model(
        self,
        project_id: str,
        model_name: str,
        version: str,
        target_stage: ModelStage,
    ) -> ModelManifest:
        """Promote model to target stage"""
        bucket_name = f"onto-{project_id}-artifacts"
        manifest_path = f"models/{model_name}-{version}/_manifest.json"

        response = self._client.get_object(bucket_name, manifest_path)
        manifest_data = json.loads(response.read())
        response.close()
        response.release_conn()

        manifest = ModelManifest(**manifest_data)
        manifest.stage = target_stage

        manifest_bytes = manifest.model_dump_json(indent=2).encode("utf-8")
        self._client.put_object(
            bucket_name=bucket_name,
            object_name=manifest_path,
            data=BytesIO(manifest_bytes),
            length=len(manifest_bytes),
            content_type="application/json",
        )

        return manifest

    def get_production_model(
        self, project_id: str, model_name: str
    ) -> ModelManifest | None:
        """Get the production-stage model"""
        bucket_name = f"onto-{project_id}-artifacts"
        prefix = f"models/{model_name}-"

        versions: list[ModelManifest] = []
        for obj in self._client.list_objects(bucket_name, prefix=prefix, recursive=True):
            if obj.object_name and obj.object_name.endswith("/_manifest.json"):
                response = self._client.get_object(bucket_name, obj.object_name)
                manifest_data = json.loads(response.read())
                response.close()
                response.release_conn()
                manifest = ModelManifest(**manifest_data)
                if manifest.stage == ModelStage.PRODUCTION:
                    versions.append(manifest)

        if not versions:
            return None
        return sorted(versions, key=lambda m: m.created_at, reverse=True)[0]

Presigned URL Upload Flow:

┌──────────┐     ┌──────────────┐     ┌──────────┐
│ Browser  │     │ coomia-dip    │     │  MinIO   │
│ (Client) │     │ API Gateway  │     │  Server  │
└────┬─────┘     └──────┬───────┘     └────┬─────┘
     │                   │                  │
     │  1. Request upload│                  │
     │     URL           │                  │
     │ ─────────────────>│                  │
     │                   │                  │
     │                   │ 2. Verify perms  │
     │                   │    Generate      │
     │                   │    Presigned URL │
     │                   │                  │
     │  3. Return        │                  │
     │     Presigned URL │                  │
     │ <─────────────────│                  │
     │                   │                  │
     │  4. PUT file to   │                  │
     │     Presigned URL │                  │
     │ ──────────────────────────────────>  │
     │                   │                  │
     │  5. 200 OK        │                  │
     │ <──────────────────────────────────  │
     │                   │                  │
     │  6. Notify upload │                  │
     │     complete      │                  │
     │ ─────────────────>│                  │
     │                   │ 7. Verify object │
     │                   │ ────────────────>│
     │                   │                  │
     │  8. Confirm done  │                  │
     │ <─────────────────│                  │

# control-Layer/api/storage_service.py
from datetime import timedelta
from urllib.parse import urlparse

from minio import Minio
from pydantic import BaseModel


class PresignedUploadResponse(BaseModel):
    """Presigned upload response"""
    upload_url: str
    object_key: str
    expires_in_seconds: int
    required_headers: dict[str, str]


class PresignedDownloadResponse(BaseModel):
    """Presigned download response"""
    download_url: str
    expires_in_seconds: int
    file_name: str
    file_size: int | None = None


class StorageService:
    """Object storage service"""

    def __init__(self, minio_client: Minio, external_endpoint: str):
        self._client = minio_client
        self._external_endpoint = external_endpoint

    def generate_upload_url(
        self,
        project_id: str,
        category: str,
        object_path: str,
        content_type: str = "application/octet-stream",
        max_size_mb: int = 500,
        expires_minutes: int = 30,
    ) -> PresignedUploadResponse:
        """Generate a presigned upload URL"""
        bucket_name = f"onto-{project_id}-{category}"
        if not self._client.bucket_exists(bucket_name):
            self._client.make_bucket(bucket_name)

        url = self._client.presigned_put_object(
            bucket_name=bucket_name,
            object_name=object_path,
            expires=timedelta(minutes=expires_minutes),
        )

        url = self._replace_endpoint(url)

        return PresignedUploadResponse(
            upload_url=url,
            object_key=f"{bucket_name}/{object_path}",
            expires_in_seconds=expires_minutes * 60,
            required_headers={
                "Content-Type": content_type,
                "x-amz-meta-project-id": project_id,
            },
        )

    def generate_download_url(
        self,
        project_id: str,
        category: str,
        object_path: str,
        expires_minutes: int = 60,
        filename_override: str | None = None,
    ) -> PresignedDownloadResponse:
        """Generate a presigned download URL"""
        bucket_name = f"onto-{project_id}-{category}"

        stat = self._client.stat_object(bucket_name, object_path)

        extra_query_params = {}
        if filename_override:
            extra_query_params["response-content-disposition"] = (
                f'attachment; filename="{filename_override}"'
            )

        url = self._client.presigned_get_object(
            bucket_name=bucket_name,
            object_name=object_path,
            expires=timedelta(minutes=expires_minutes),
            extra_query_params=extra_query_params if extra_query_params else None,
        )

        url = self._replace_endpoint(url)

        return PresignedDownloadResponse(
            download_url=url,
            expires_in_seconds=expires_minutes * 60,
            file_name=filename_override or object_path.rsplit("/", 1)[-1],
            file_size=stat.size,
        )

    def _replace_endpoint(self, url: str) -> str:
        """Replace internal endpoint with externally-accessible endpoint"""
        parsed = urlparse(url)
        return url.replace(
            f"{parsed.scheme}://{parsed.netloc}",
            self._external_endpoint,
        )

// Frontend upload example (TypeScript)
interface PresignedUploadResponse {
  upload_url: string;
  object_key: string;
  expires_in_seconds: number;
  required_headers: Record<string, string>;
}

async function uploadFileToMinIO(
  file: File,
  projectId: string,
  category: string,
): Promise<string> {
  // Step 1: Get presigned URL from coomia-dip API
  const response = await fetch('/api/v1/storage/upload-url', {
    method: 'POST',
    headers: { 'Content-Type': 'application/json' },
    body: JSON.stringify({
      project_id: projectId,
      category: category,
      object_path: `uploads/${Date.now()}-${file.name}`,
      content_type: file.type,
    }),
  });
  const presigned: PresignedUploadResponse = await response.json();

  // Step 2: Upload directly to MinIO (bypasses app server)
  const uploadResponse = await fetch(presigned.upload_url, {
    method: 'PUT',
    headers: {
      ...presigned.required_headers,
      'Content-Type': file.type,
    },
    body: file,
  });

  if (!uploadResponse.ok) {
    throw new Error(`Upload failed: ${uploadResponse.statusText}`);
  }

  // Step 3: Notify app server of completion
  await fetch('/api/v1/storage/upload-complete', {
    method: 'POST',
    headers: { 'Content-Type': 'application/json' },
    body: JSON.stringify({ object_key: presigned.object_key }),
  });

  return presigned.object_key;
}

# deployment-Layer/scripts/minio_lifecycle_setup.py
"""MinIO lifecycle policy configuration script"""

from minio import Minio
from minio.lifecycleconfig import (
    LifecycleConfig,
    Rule,
    Filter,
    Expiration,
)


def setup_lifecycle_policies(client: Minio, project_id: str) -> None:
    """Configure project lifecycle policies"""

    # Pipeline bucket: logs expire in 30 days, outputs in 90 days
    pipeline_bucket = f"onto-{project_id}-pipeline"
    if client.bucket_exists(pipeline_bucket):
        config = LifecycleConfig(
            [
                Rule(
                    rule_id="expire-logs-30d",
                    status="Enabled",
                    rule_filter=Filter(prefix="runs/*/logs/"),
                    expiration=Expiration(days=30),
                ),
                Rule(
                    rule_id="expire-outputs-90d",
                    status="Enabled",
                    rule_filter=Filter(prefix="runs/*/output/"),
                    expiration=Expiration(days=90),
                ),
                Rule(
                    rule_id="keep-snapshots",
                    status="Enabled",
                    rule_filter=Filter(prefix="snapshots/"),
                    expiration=Expiration(days=365),
                ),
            ]
        )
        client.set_bucket_lifecycle(pipeline_bucket, config)

    # Artifacts bucket: checkpoints expire in 14 days, models retained
    artifacts_bucket = f"onto-{project_id}-artifacts"
    if client.bucket_exists(artifacts_bucket):
        config = LifecycleConfig(
            [
                Rule(
                    rule_id="expire-checkpoints-14d",
                    status="Enabled",
                    rule_filter=Filter(prefix="checkpoints/"),
                    expiration=Expiration(days=14),
                ),
                Rule(
                    rule_id="expire-evaluations-180d",
                    status="Enabled",
                    rule_filter=Filter(prefix="evaluations/"),
                    expiration=Expiration(days=180),
                ),
            ]
        )
        client.set_bucket_lifecycle(artifacts_bucket, config)

    # Exports bucket: all files expire in 7 days
    exports_bucket = f"onto-{project_id}-exports"
    if client.bucket_exists(exports_bucket):
        config = LifecycleConfig(
            [
                Rule(
                    rule_id="expire-exports-7d",
                    status="Enabled",
                    rule_filter=Filter(prefix=""),
                    expiration=Expiration(days=7),
                ),
            ]
        )
        client.set_bucket_lifecycle(exports_bucket, config)

# python-sdk/ontology_sdk/storage/usage_monitor.py
from dataclasses import dataclass
from minio import Minio


@dataclass
class BucketUsage:
    """Bucket usage statistics"""
    bucket_name: str
    object_count: int
    total_size_bytes: int
    largest_object_bytes: int
    largest_object_name: str

    @property
    def total_size_mb(self) -> float:
        return self.total_size_bytes / (1024 * 1024)

    @property
    def total_size_gb(self) -> float:
        return self.total_size_bytes / (1024 * 1024 * 1024)


class UsageMonitor:
    """Storage usage monitor"""

    def __init__(self, minio_client: Minio):
        self._client = minio_client

    def get_bucket_usage(self, bucket_name: str) -> BucketUsage:
        """Get usage for a single bucket"""
        total_size = 0
        count = 0
        largest_size = 0
        largest_name = ""

        for obj in self._client.list_objects(bucket_name, recursive=True):
            count += 1
            size = obj.size or 0
            total_size += size
            if size > largest_size:
                largest_size = size
                largest_name = obj.object_name or ""

        return BucketUsage(
            bucket_name=bucket_name,
            object_count=count,
            total_size_bytes=total_size,
            largest_object_bytes=largest_size,
            largest_object_name=largest_name,
        )

    def get_project_usage(self, project_id: str) -> list[BucketUsage]:
        """Get usage for all project buckets"""
        results: list[BucketUsage] = []
        for bucket in self._client.list_buckets():
            if bucket.name and bucket.name.startswith(f"onto-{project_id}-"):
                results.append(self.get_bucket_usage(bucket.name))
        return results

    def check_quota(
        self, project_id: str, quota_gb: float = 100.0
    ) -> tuple[bool, float]:
        """Check if project is within quota"""
        usages = self.get_project_usage(project_id)
        total_gb = sum(u.total_size_gb for u in usages)
        return total_gb <= quota_gb, total_gb

-- Object reference fields in entity_common
CREATE TABLE entity_common (
    entity_id       VARCHAR(64) NOT NULL,
    entity_type     VARCHAR(128) NOT NULL,
    display_name    VARCHAR(512),
    properties      JSON,
    -- Object references stored in properties
    -- {"model_artifact": "s3://onto-proj001-artifacts/models/risk-v1/model.onnx"}
    -- {"report_file": "s3://onto-proj001-exports/2024-06/monthly.pdf"}
    created_at      DATETIME NOT NULL,
    updated_at      DATETIME NOT NULL
);

# python-sdk/ontology_sdk/storage/object_ref.py
import re
from dataclasses import dataclass


@dataclass
class ObjectReference:
    """S3 object reference"""
    bucket: str
    key: str
    version_id: str | None = None

    @classmethod
    def parse(cls, uri: str) -> "ObjectReference":
        """Parse s3:// URI"""
        pattern = r"^s3://([^/]+)/(.+?)(?:\?versionId=(.+))?$"
        match = re.match(pattern, uri)
        if not match:
            raise ValueError(f"Invalid S3 URI: {uri}")
        return cls(
            bucket=match.group(1),
            key=match.group(2),
            version_id=match.group(3),
        )

    def to_uri(self) -> str:
        """Convert to s3:// URI"""
        base = f"s3://{self.bucket}/{self.key}"
        if self.version_id:
            base += f"?versionId={self.version_id}"
        return base

    @property
    def project_id(self) -> str:
        """Extract project ID from bucket name"""
        parts = self.bucket.split("-")
        if len(parts) >= 3 and parts[0] == "onto":
            return parts[1]
        raise ValueError(f"Cannot extract project_id from bucket: {self.bucket}")

    @property
    def category(self) -> str:
        """Extract category from bucket name"""
        parts = self.bucket.split("-")
        if len(parts) >= 3 and parts[0] == "onto":
            return "-".join(parts[2:])
        raise ValueError(f"Cannot extract category from bucket: {self.bucket}")

Multipart Upload Flow:

┌──────────┐                              ┌──────────┐
│  Client  │                              │  MinIO   │
└────┬─────┘                              └────┬─────┘
     │  1. InitiateMultipartUpload              │
     │ ────────────────────────────────────────>│
     │                                          │
     │  2. UploadId                             │
     │ <────────────────────────────────────────│
     │                                          │
     │  3. UploadPart (Part 1: 0-100MB)         │
     │ ────────────────────────────────────────>│
     │  4. ETag for Part 1                      │
     │ <────────────────────────────────────────│
     │                                          │
     │  5. UploadPart (Part 2: 100-200MB)       │
     │ ────────────────────────────────────────>│
     │  6. ETag for Part 2                      │
     │ <────────────────────────────────────────│
     │                                          │
     │  ... (parallel part uploads)             │
     │                                          │
     │  N. CompleteMultipartUpload               │
     │     [Part1:ETag1, Part2:ETag2, ...]      │
     │ ────────────────────────────────────────>│
     │                                          │
     │  N+1. Object Created                     │
     │ <────────────────────────────────────────│

# python-sdk/ontology_sdk/storage/multipart_upload.py
import hashlib
from dataclasses import dataclass
from pathlib import Path
from typing import Callable

from minio import Minio


PART_SIZE = 100 * 1024 * 1024  # 100 MB per part


@dataclass
class UploadProgress:
    """Upload progress tracker"""
    total_bytes: int
    uploaded_bytes: int = 0
    parts_completed: int = 0
    total_parts: int = 0

    @property
    def percentage(self) -> float:
        if self.total_bytes == 0:
            return 100.0
        return (self.uploaded_bytes / self.total_bytes) * 100


@dataclass
class MultipartUploader:
    """Large file multipart uploader"""
    minio_client: Minio
    max_workers: int = 4
    part_size: int = PART_SIZE

    def upload_large_file(
        self,
        bucket_name: str,
        object_name: str,
        file_path: str | Path,
        content_type: str = "application/octet-stream",
        progress_callback: Callable[[UploadProgress], None] | None = None,
    ) -> str:
        """Upload large file with multipart, returns ETag"""
        file_path = Path(file_path)
        file_size = file_path.stat().st_size

        if file_size <= self.part_size:
            self.minio_client.fput_object(
                bucket_name, object_name, str(file_path),
                content_type=content_type,
            )
            return hashlib.md5(file_path.read_bytes()).hexdigest()

        total_parts = (file_size + self.part_size - 1) // self.part_size
        progress = UploadProgress(
            total_bytes=file_size,
            total_parts=total_parts,
        )

        result = self.minio_client.fput_object(
            bucket_name,
            object_name,
            str(file_path),
            content_type=content_type,
            part_size=self.part_size,
        )

        progress.uploaded_bytes = file_size
        progress.parts_completed = total_parts
        if progress_callback:
            progress_callback(progress)

        return result.etag or ""

# MinIO server-side performance tuning

# Enable async writes (higher throughput, lower latency)
export MINIO_DRIVE_SYNC=off

# Reduce background I/O impact
export MINIO_SCANNER_SPEED=slow

# Concurrency settings
export MINIO_API_REQUESTS_MAX=1600
export MINIO_API_REQUESTS_DEADLINE=10s

# Cache settings
export MINIO_CACHE_DRIVES="/mnt/cache1,/mnt/cache2"
export MINIO_CACHE_QUOTA=80
export MINIO_CACHE_AFTER=3
export MINIO_CACHE_WATERMARK_LOW=70
export MINIO_CACHE_WATERMARK_HIGH=90

# python-sdk/ontology_sdk/storage/best_practices.py
"""MinIO client best practices"""

from minio import Minio
import urllib3


def create_optimized_client(
    endpoint: str,
    access_key: str,
    secret_key: str,
    secure: bool = True,
) -> Minio:
    """Create an optimized MinIO client"""
    http_client = urllib3.PoolManager(
        num_pools=10,
        maxsize=10,
        retries=urllib3.Retry(
            total=3,
            backoff_factor=0.2,
            status_forcelist=[500, 502, 503, 504],
        ),
        timeout=urllib3.Timeout(connect=5.0, read=30.0),
    )

    return Minio(
        endpoint=endpoint,
        access_key=access_key,
        secret_key=secret_key,
        secure=secure,
        http_client=http_client,
    )


def batch_delete_objects(
    client: Minio,
    bucket_name: str,
    prefix: str,
) -> int:
    """Batch delete objects (10x faster than one-by-one)"""
    from minio.deleteobjects import DeleteObject

    delete_list = [
        DeleteObject(obj.object_name)
        for obj in client.list_objects(bucket_name, prefix=prefix, recursive=True)
        if obj.object_name
    ]

    if not delete_list:
        return 0

    errors = list(client.remove_objects(bucket_name, delete_list))
    if errors:
        for err in errors:
            print(f"Delete error: {err}")

    return len(delete_list) - len(errors)