Add opensearch

.github/workflows/cicd.yml

@@ -70,11 +70,11 @@ jobs:
 
       - name: Install elasticsearch stac-fastapi
         run: |
-          pip install ./stac_fastapi/elasticsearch[dev,server]
+          pip install ./stac_fastapi/elastic_search[dev,server]
 
       - name: Run test suite against Elasticsearch 7.x
         run: |
-          cd stac_fastapi/elasticsearch && pipenv run pytest -svvv
+          cd stac_fastapi/elastic_search && pipenv run pytest -svvv
         env:
           ENVIRONMENT: testing
           ES_PORT: 9200
@@ -84,7 +84,7 @@ jobs:
 
       - name: Run test suite against Elasticsearch 8.x
         run: |
-          cd stac_fastapi/elasticsearch && pipenv run pytest -svvv
+          cd stac_fastapi/elastic_search && pipenv run pytest -svvv
         env:
           ENVIRONMENT: testing
           ES_PORT: 9400

.pre-commit-config.yaml

@@ -16,7 +16,7 @@ repos:
           # E501 let black handle all line length decisions
           # W503 black conflicts with "line break before operator" rule
           # E203 black conflicts with "whitespace before ':'" rule
-          '--ignore=E501,W503,E203,C901' ]
+          '--ignore=E501,W503,E203,C901,E402' ]
   - repo: https://github.com/pre-commit/mirrors-mypy
     rev: v0.991
     hooks:

Dockerfile.dev.es

@@ -0,0 +1,20 @@
+FROM python:3.10-slim
+
+# update apt pkgs, and install build-essential for ciso8601
+RUN apt-get update && \
+    apt-get -y upgrade && \
+    apt-get install -y build-essential && \
+    apt-get clean && \
+    rm -rf /var/lib/apt/lists/*
+
+# update certs used by Requests
+ENV CURL_CA BUNDLE=/etc/ssl/certs/ca-certificates.crt
+
+WORKDIR /app
+
+# Copy the contents of common and elastic_search directories directly into /app
+COPY ./stac_fastapi/common /app/stac_fastapi/common
+COPY ./stac_fastapi/elastic_search /app/stac_fastapi/elastic_search
+
+# Install dependencies
+RUN pip install --no-cache-dir -e ./stac_fastapi/elastic_search[dev,server]

docker-compose.yml

@@ -3,11 +3,11 @@ version: '3.9'
 services:
   app-elasticsearch:
     container_name: stac-fastapi-es
-    image: stac-utils/stac-fastapi
+    image: stac-utils/stac-fastapi-es
     restart: always
     build:
       context: .
-      dockerfile: Dockerfile.dev
+      dockerfile: Dockerfile.dev.es
     environment:
       - APP_HOST=0.0.0.0
       - APP_PORT=8080
@@ -18,16 +18,47 @@ services:
       - ES_PORT=9200
       - ES_USE_SSL=false
       - ES_VERIFY_CERTS=false
+      - DB_TYPE=elasticsearch
     ports:
       - "8080:8080"
     volumes:
-      - ./stac_fastapi:/app/stac_fastapi
+      - ./stac_fastapi/elastic_search:/app/stac_fastapi/elastic_search
+      - ./stac_fastapi/common:/app/stac_fastapi/common
       - ./scripts:/app/scripts
       - ./esdata:/usr/share/elasticsearch/data
     depends_on:
       - elasticsearch
     command:
-      bash -c "./scripts/wait-for-it-es.sh es-container:9200 && python -m stac_fastapi.elasticsearch.app"
+      bash -c "./scripts/wait-for-it-es.sh es-container:9200 && python -m stac_fastapi.elastic_search.app"
+
+  app-opensearch:
+    container_name: stac-fastapi-os
+    image: stac-utils/stac-fastapi-os
+    restart: always
+    build:
+      context: .
+      dockerfile: Dockerfile.dev
+    environment:
+      - APP_HOST=0.0.0.0
+      - APP_PORT=8080
+      - RELOAD=true
+      - ENVIRONMENT=local
+      - WEB_CONCURRENCY=10
+      - ES_HOST=172.17.0.1
+      - ES_PORT=9200
+      - ES_USE_SSL=false
+      - ES_VERIFY_CERTS=false
+      - DB_TYPE=opensearch
+    ports:
+      - "8082:8080"
+    volumes:
+      - ./stac_fastapi:/app/stac_fastapi
+      - ./scripts:/app/scripts
+      - ./osdata:/usr/share/opensearch/data
+    depends_on:
+      - opensearch
+    command:
+      bash -c "./scripts/wait-for-it-es.sh os-container:9200 && python -m stac_fastapi.elastic_search.app"
 
   elasticsearch:
     container_name: es-container
@@ -39,3 +70,15 @@ services:
       - ./elasticsearch/snapshots:/usr/share/elasticsearch/snapshots
     ports:
       - "9200:9200"
+
+  opensearch:
+    container_name: os-container
+    image: opensearchproject/opensearch:latest
+    environment:
+      - "discovery.type=single-node"
+      - "plugins.security.disabled=true"
+    volumes:
+      - ./opensearch/config/opensearch.yml:/usr/share/opensearch/config/opensearch.yml
+      - ./opensearch/snapshots:/usr/share/opensearch/snapshots
+    ports:
+      - "9202:9200"

examples/pip_docker/docker-compose.yml

@@ -27,7 +27,7 @@ services:
     depends_on:
       - elasticsearch
     command:
-      bash -c "./scripts/wait-for-it-es.sh es-container:9200 && python -m stac_fastapi.elasticsearch.app"
+      bash -c "./scripts/wait-for-it-es.sh es-container:9200 && python -m stac_fastapi.elastic_search.app"
 
   elasticsearch:
     container_name: es-container

opensearch/config/opensearch.yml

@@ -0,0 +1,35 @@
+## Cluster Settings
+cluster.name: stac-cluster
+node.name: es01
+network.host: 0.0.0.0
+transport.host: 0.0.0.0
+discovery.type: single-node
+http.port: 9200
+
+path:
+  repo:
+    - /usr/share/opensearch/snapshots
+
+######## Start OpenSearch Security Demo Configuration ########
+# WARNING: revise all the lines below before you go into production
+plugins.security.ssl.transport.pemcert_filepath: esnode.pem
+plugins.security.ssl.transport.pemkey_filepath: esnode-key.pem
+plugins.security.ssl.transport.pemtrustedcas_filepath: root-ca.pem
+plugins.security.ssl.transport.enforce_hostname_verification: false
+plugins.security.ssl.http.enabled: true
+plugins.security.ssl.http.pemcert_filepath: esnode.pem
+plugins.security.ssl.http.pemkey_filepath: esnode-key.pem
+plugins.security.ssl.http.pemtrustedcas_filepath: root-ca.pem
+plugins.security.allow_unsafe_democertificates: true
+plugins.security.allow_default_init_securityindex: true
+plugins.security.authcz.admin_dn:
+  - CN=kirk,OU=client,O=client,L=test, C=de
+
+plugins.security.audit.type: internal_opensearch
+plugins.security.enable_snapshot_restore_privilege: true
+plugins.security.check_snapshot_restore_write_privileges: true
+plugins.security.restapi.roles_enabled: ["all_access", "security_rest_api_access"]
+plugins.security.system_indices.enabled: true
+plugins.security.system_indices.indices: [".plugins-ml-config", ".plugins-ml-connector", ".plugins-ml-model-group", ".plugins-ml-model", ".plugins-ml-task", ".plugins-ml-conversation-meta", ".plugins-ml-conversation-interactions", ".opendistro-alerting-config", ".opendistro-alerting-alert*", ".opendistro-anomaly-results*", ".opendistro-anomaly-detector*", ".opendistro-anomaly-checkpoints", ".opendistro-anomaly-detection-state", ".opendistro-reports-*", ".opensearch-notifications-*", ".opensearch-notebooks", ".opensearch-observability", ".ql-datasources", ".opendistro-asynchronous-search-response*", ".replication-metadata-store", ".opensearch-knn-models", ".geospatial-ip2geo-data*"]
+node.max_local_storage_nodes: 3
+######## End OpenSearch Security Demo Configuration ########

stac_fastapi/common/__init__.py

@@ -0,0 +1 @@
+"""elasticsearch/ opensearch shared code."""

stac_fastapi/common/base_database_logic.py

@@ -0,0 +1,47 @@
+"""base class for database logic."""
+
+# from abc import ABC, abstractmethod
+# from typing import Any, Dict, List, Optional, Tuple, Union
+
+# from stac_fastapi.types.stac import Collection, Item
+
+
+# class BaseDatabaseLogic(ABC):
+#     """
+#     Abstract base class for database logic.
+#     This class defines the interface for database operations.
+#     """
+
+#     @abstractmethod
+#     async def get_all_collections(self, token: Optional[str], limit: int) -> Any:
+#         pass
+
+#     @abstractmethod
+#     async def get_one_item(self, collection_id: str, item_id: str) -> Dict:
+#         pass
+
+#     @abstractmethod
+#     async def create_item(self, item: Item, refresh: bool = False) -> None:
+#         pass
+
+#     @abstractmethod
+#     async def delete_item(
+#         self, item_id: str, collection_id: str, refresh: bool = False
+#     ) -> None:
+#         pass
+
+#     @abstractmethod
+#     async def create_collection(
+#         self, collection: Collection, refresh: bool = False
+#     ) -> None:
+#         pass
+
+#     @abstractmethod
+#     async def find_collection(self, collection_id: str) -> Collection:
+#         pass
+
+#     @abstractmethod
+#     async def delete_collection(
+#         self, collection_id: str, refresh: bool = False
+#     ) -> None:
+#         pass

stac_fastapi/elasticsearch/stac_fastapi/elasticsearch/core.py → stac_fastapi/common/core.py

@@ -1,4 +1,5 @@
 """Item crud client."""
+import importlib
 import logging
 import re
 from base64 import urlsafe_b64encode
@@ -18,12 +19,11 @@
 from stac_pydantic.links import Relations
 from stac_pydantic.shared import MimeTypes
 
-from stac_fastapi.elasticsearch import serializers
-from stac_fastapi.elasticsearch.config import ElasticsearchSettings
-from stac_fastapi.elasticsearch.database_logic import DatabaseLogic
-from stac_fastapi.elasticsearch.models.links import PagingLinks
-from stac_fastapi.elasticsearch.serializers import CollectionSerializer, ItemSerializer
-from stac_fastapi.elasticsearch.session import Session
+# from common.base_database_logic import BaseDatabaseLogic
+from stac_fastapi.common.models.links import PagingLinks
+from stac_fastapi.elastic_search import serializers
+from stac_fastapi.elastic_search.serializers import CollectionSerializer, ItemSerializer
+from stac_fastapi.elastic_search.session import Session
 from stac_fastapi.extensions.third_party.bulk_transactions import (
     BaseBulkTransactionsClient,
     BulkTransactionMethod,
@@ -66,12 +66,36 @@ class CoreClient(AsyncBaseCoreClient):
 
     session: Session = attr.ib(default=attr.Factory(Session.create_from_env))
     item_serializer: Type[serializers.ItemSerializer] = attr.ib(
-        default=serializers.ItemSerializer
+        default=serializers.ItemSerializer  # type: ignore
     )
     collection_serializer: Type[serializers.CollectionSerializer] = attr.ib(
-        default=serializers.CollectionSerializer
+        default=serializers.CollectionSerializer  # type: ignore
     )
-    database = DatabaseLogic()
+    post_request_model = attr.ib(default=BaseSearchPostRequest)
+
+    def __attrs_post_init__(self):
+        """
+        Post-initialization method for CoreClient.
+
+        This method is automatically called after CoreClient's instance is initialized.
+        It's responsible for setting up the database logic dynamically based on the
+        environment or configuration, ensuring that the appropriate database logic
+        (Elasticsearch or OpenSearch) is used.
+        """
+        try:
+            # Dynamically import the database logic based on installed package
+            database_module = importlib.import_module(
+                "stac_fastapi.elastic_search.database_logic"
+            )
+            DatabaseLogicClass = getattr(database_module, "DatabaseLogic")
+        except ImportError:
+            # Fall back to OpenSearch if Elasticsearch is not available
+            database_module = importlib.import_module(
+                "stac_fastapi.open_search.database_logic"
+            )
+            DatabaseLogicClass = getattr(database_module, "DatabaseLogic")
+
+        self.database = DatabaseLogicClass()
 
     @overrides
     async def all_collections(self, **kwargs) -> Collections:
@@ -542,7 +566,31 @@ class TransactionsClient(AsyncBaseTransactionsClient):
     """Transactions extension specific CRUD operations."""
 
     session: Session = attr.ib(default=attr.Factory(Session.create_from_env))
-    database = DatabaseLogic()
+    # database: BaseDatabaseLogic = attr.ib(init=False)
+
+    def __attrs_post_init__(self):
+        """
+        Post-initialization method for TransactionsClient.
+
+        This method is automatically called after the instance is initialized.
+        It's responsible for setting up the database logic dynamically based on the
+        environment or configuration, ensuring that the appropriate database logic
+        (Elasticsearch or OpenSearch) is used.
+        """
+        try:
+            # Dynamically import the database logic based on installed package
+            database_module = importlib.import_module(
+                "stac_fastapi.elastic_search.database_logic"
+            )
+            DatabaseLogicClass = getattr(database_module, "DatabaseLogic")
+        except ImportError:
+            # Fall back to OpenSearch if Elasticsearch is not available
+            database_module = importlib.import_module(
+                "stac_fastapi.opensearch.database_logic"
+            )
+            DatabaseLogicClass = getattr(database_module, "DatabaseLogic")
+
+        self.database = DatabaseLogicClass()
 
     @overrides
     async def create_item(
@@ -712,12 +760,31 @@ class BulkTransactionsClient(BaseBulkTransactionsClient):
     """
 
     session: Session = attr.ib(default=attr.Factory(Session.create_from_env))
-    database = DatabaseLogic()
+    # database: BaseDatabaseLogic = attr.ib(init=False)
 
     def __attrs_post_init__(self):
-        """Create es engine."""
-        settings = ElasticsearchSettings()
-        self.client = settings.create_client
+        """
+        Post-initialization method for BulkTransactionsClient.
+
+        This method is automatically called after the instance is initialized.
+        It's responsible for setting up the database logic dynamically based on the
+        environment or configuration, ensuring that the appropriate database logic
+        (Elasticsearch or OpenSearch) is used.
+        """
+        try:
+            # Dynamically import the database logic based on installed package
+            database_module = importlib.import_module(
+                "stac_fastapi.elastic_search.database_logic"
+            )
+            DatabaseLogicClass = getattr(database_module, "DatabaseLogic")
+        except ImportError:
+            # Fall back to OpenSearch if Elasticsearch is not available
+            database_module = importlib.import_module(
+                "stac_fastapi.opensearch.database_logic"
+            )
+            DatabaseLogicClass = getattr(database_module, "DatabaseLogic")
+
+        self.database = DatabaseLogicClass()
 
     def preprocess_item(
         self, item: stac_types.Item, base_url, method: BulkTransactionMethod