Merge pull request #13 from jonhealy1/dev

Merge dev

Author: jonhealy1

.github/workflows/cicd.yaml

@@ -1,9 +1,9 @@
 name: stac-fastapi-nosql
 on:
   push:
-    branches: [ main ]
+    branches: [ main, dev ]
   pull_request:
-    branches: [ main ]
+    branches: [ main, dev ]
 
 jobs:
   test:

.github/workflows/publish-to-pypi.yml

@@ -0,0 +1,38 @@
+name: Publish Python distributions to PyPI
+
+on:
+  push:
+    branches:
+      - main
+
+jobs:
+  build-n-publish:
+    name: Build and publish Python distributions to PyPI and TestPyPI
+    runs-on: ubuntu-18.04
+
+    steps:
+    - uses: actions/checkout@master
+    - name: Set up Python 3.8
+      uses: actions/setup-python@v1
+      with:
+        python-version: 3.8
+
+    - name: Install pypa/build
+      run: >-
+        python -m
+        pip install
+        build
+        --user
+    - name: Build a binary wheel and a source tarball
+      run: >-
+        python -m
+        build
+        --sdist
+        --wheel
+        --outdir dist/
+        stac_fastapi/mongo
+    - name: Publish distribution to PyPI
+      # if: startsWith(github.ref, 'refs/tags')
+      uses: pypa/gh-action-pypi-publish@v1.4.2
+      with:
+        password: ${{ secrets.PYPI_API_TOKEN }}

.github/workflows/publish-to-test-pypi.yml

@@ -0,0 +1,39 @@
+name: Publish Python distributions to PyPI
+
+on:
+  push:
+    branches:
+      - dev
+
+jobs:
+  build-n-publish:
+    name: Build and publish Python distributions to PyPI and TestPyPI
+    runs-on: ubuntu-18.04
+
+    steps:
+    - uses: actions/checkout@master
+    - name: Set up Python 3.8
+      uses: actions/setup-python@v1
+      with:
+        python-version: 3.8
+
+    - name: Install pypa/build
+      run: >-
+        python -m
+        pip install
+        build
+        --user
+    - name: Build a binary wheel and a source tarball
+      run: >-
+        python -m
+        build
+        --sdist
+        --wheel
+        --outdir dist/
+        stac_fastapi/mongo
+    - name: Publish distribution to Test PyPI
+      # if: startsWith(github.ref, 'refs/tags')
+      uses: pypa/gh-action-pypi-publish@v1.4.2
+      with:
+        password: ${{ secrets.TEST_PYPI_API_TOKEN }}
+        repository_url: https://test.pypi.org/legacy/

VERSION

@@ -0,0 +1 @@
+2.3.0

stac_fastapi/elasticsearch/setup.py

@@ -16,6 +16,7 @@
     "elasticsearch[async]",
     "elasticsearch-dsl",
     "pystac[validation]",
+    "uvicorn",
 ]
 
 extra_reqs = {
@@ -47,7 +48,7 @@
     keywords="STAC FastAPI COG",
     author=u"Arturo Engineering",
     author_email="engineering@arturo.ai",
-    url="https://github.com/stac-utils/stac-fastapi",
+    url="https://github.com/jonhealy1/stac-fastapi-nosql",
     license="MIT",
     packages=find_namespace_packages(exclude=["alembic", "tests", "scripts"]),
     zip_safe=False,

stac_fastapi/elasticsearch/stac_fastapi/elasticsearch/core.py

@@ -94,7 +94,7 @@ def get_collection(self, collection_id: str, **kwargs) -> Collection:
         return self.collection_serializer.db_to_stac(collection["_source"], base_url)
 
     def item_collection(
-        self, collection_id: str, limit: int = 10, token: str = None, **kwargs
+        self, collection_id: str, limit: int = 10, **kwargs
     ) -> ItemCollection:
         """Read an item collection from the database."""
         links = []

stac_fastapi/elasticsearch/stac_fastapi/elasticsearch/serializers.py

@@ -37,15 +37,17 @@ def db_to_stac(cls, item: dict, base_url: str) -> stac_types.Item:
 
         return stac_types.Item(
             type="Feature",
-            stac_version=item["stac_version"],
-            stac_extensions=item["stac_extensions"] or [],
-            id=item["id"],
-            collection=item["collection"],
-            geometry=item["geometry"],
-            bbox=item["bbox"],
-            properties=item["properties"],
-            links=item_links,
-            assets=item["assets"],
+            stac_version=item["stac_version"] if "stac_version" in item else "",
+            stac_extensions=item["stac_extensions"]
+            if "stac_extensions" in item
+            else [],
+            id=item_id,
+            collection=item["collection"] if "collection" in item else "",
+            geometry=item["geometry"] if "geometry" in item else {},
+            bbox=item["bbox"] if "bbox" in item else [],
+            properties=item["properties"] if "properties" in item else {},
+            links=item_links if "links" in item else [],
+            assets=item["assets"] if "assets" in item else {},
         )
 
 
@@ -66,14 +68,20 @@ def db_to_stac(cls, collection: dict, base_url: str) -> stac_types.Collection:
         return stac_types.Collection(
             type="Collection",
             id=collection["id"],
-            stac_extensions=collection["stac_extensions"] or [],
-            stac_version=collection["stac_version"],
-            title=collection["title"],
-            description=collection["description"],
-            keywords=collection["keywords"],
-            license=collection["license"],
-            providers=collection["providers"],
-            summaries=collection["summaries"],
-            extent=collection["extent"],
+            stac_extensions=collection["stac_extensions"]
+            if "stac_extensions" in collection
+            else [],
+            stac_version=collection["stac_version"]
+            if "stac_version" in collection
+            else "",
+            title=collection["title"] if "title" in collection else "",
+            description=collection["description"]
+            if "description" in collection
+            else "",
+            keywords=collection["keywords"] if "keywords" in collection else [],
+            license=collection["license"] if "license" in collection else "",
+            providers=collection["providers"] if "providers" in collection else {},
+            summaries=collection["summaries"] if "summaries" in collection else {},
+            extent=collection["extent"] if "extent" in collection else {},
             links=collection_links,
         )

stac_fastapi/elasticsearch/stac_fastapi/elasticsearch/transactions.py

@@ -31,6 +31,26 @@ class TransactionsClient(BaseTransactionsClient):
     settings = ElasticsearchSettings()
     client = settings.create_client
 
+    def _create_item_index(self):
+        mapping = {
+            "mappings": {
+                "properties": {
+                    "geometry": {"type": "geo_shape"},
+                    "id": {"type": "text", "fields": {"keyword": {"type": "keyword"}}},
+                    "properties__datetime": {
+                        "type": "text",
+                        "fields": {"keyword": {"type": "keyword"}},
+                    },
+                }
+            }
+        }
+
+        _ = self.client.indices.create(
+            index="stac_items",
+            body=mapping,
+            ignore=400,  # ignore 400 already exists code
+        )
+
     def create_item(self, model: stac_types.Item, **kwargs):
         """Create item."""
         base_url = str(kwargs["request"].base_url)
@@ -59,24 +79,7 @@ def create_item(self, model: stac_types.Item, **kwargs):
         if "created" not in model["properties"]:
             model["properties"]["created"] = str(now)
 
-        mapping = {
-            "mappings": {
-                "properties": {
-                    "geometry": {"type": "geo_shape"},
-                    "id": {"type": "text", "fields": {"keyword": {"type": "keyword"}}},
-                    "properties__datetime": {
-                        "type": "text",
-                        "fields": {"keyword": {"type": "keyword"}},
-                    },
-                }
-            }
-        }
-
-        _ = self.client.indices.create(
-            index="stac_items",
-            body=mapping,
-            ignore=400,  # ignore 400 already exists code
-        )
+        self._create_item_index()
 
         self.client.index(
             index="stac_items", doc_type="_doc", id=model["id"], document=model
@@ -91,6 +94,8 @@ def create_collection(self, model: stac_types.Collection, **kwargs):
         ).create_links()
         model["links"] = collection_links
 
+        self._create_item_index()
+
         if self.client.exists(index="stac_collections", id=model["id"]):
             raise ConflictError(f"Collection {model['id']} already exists")
         self.client.index(
@@ -155,39 +160,79 @@ def __attrs_post_init__(self):
         settings = ElasticsearchSettings()
         self.client = settings.create_client
 
+    def _create_item_index(self):
+        mapping = {
+            "mappings": {
+                "properties": {
+                    "geometry": {"type": "geo_shape"},
+                    "id": {"type": "text", "fields": {"keyword": {"type": "keyword"}}},
+                    "properties__datetime": {
+                        "type": "text",
+                        "fields": {"keyword": {"type": "keyword"}},
+                    },
+                }
+            }
+        }
+
+        _ = self.client.indices.create(
+            index="stac_items",
+            body=mapping,
+            ignore=400,  # ignore 400 already exists code
+        )
+
     def _preprocess_item(self, model: stac_types.Item, base_url) -> stac_types.Item:
         """Preprocess items to match data model."""
         item_links = ItemLinks(
             collection_id=model["collection"], item_id=model["id"], base_url=base_url
         ).create_links()
         model["links"] = item_links
 
-        # with self.client.start_session(causal_consistency=True) as session:
-        #     error_check = ErrorChecks(session=session, client=self.client)
-        #     error_check._check_collection_foreign_key(model)
-        #     error_check._check_item_conflict(model)
-        #     now = datetime.utcnow().strftime(DATETIME_RFC339)
-        #     if "created" not in model["properties"]:
-        #         model["properties"]["created"] = str(now)
-        #     return model
+        if not self.client.exists(index="stac_collections", id=model["collection"]):
+            raise ForeignKeyError(f"Collection {model['collection']} does not exist")
+
+        if self.client.exists(index="stac_items", id=model["id"]):
+            raise ConflictError(
+                f"Item {model['id']} in collection {model['collection']} already exists"
+            )
+
+        now = datetime.utcnow().strftime(DATETIME_RFC339)
+        if "created" not in model["properties"]:
+            model["properties"]["created"] = str(now)
+
+        # elasticsearch doesn't like the fact that some values are float and some were int
+        if "eo:bands" in model["properties"]:
+            for wave in model["properties"]["eo:bands"]:
+                for k, v in wave.items():
+                    if type(v) != str:
+                        v = float(v)
+                        wave.update({k: v})
+        return model
 
     def bulk_item_insert(self, items: Items, **kwargs) -> str:
         """Bulk item insertion using es."""
+        self._create_item_index()
         try:
             base_url = str(kwargs["request"].base_url)
         except Exception:
             base_url = ""
         processed_items = [self._preprocess_item(item, base_url) for item in items]
         return_msg = f"Successfully added {len(processed_items)} items."
-        # with self.client.start_session(causal_consistency=True) as session:
-        #     self.item_table.insert_many(processed_items, session=session)
-        #     return return_msg
 
-        helpers.bulk(
-            self.client,
-            processed_items,
-            index="stac_items",
-            doc_type="_doc",
-            request_timeout=200,
-        )
+        # helpers.bulk(
+        #     self.client,
+        #     processed_items,
+        #     index="stac_items",
+        #     doc_type="_doc",
+        #     request_timeout=200,
+        # )
+
+        def bulk_sync(processed_items):
+            actions = [
+                {"_index": "stac_items", "_source": item} for item in processed_items
+            ]
+
+            helpers.bulk(self.client, actions)
+
+        bulk_sync(processed_items)
+
         return return_msg

stac_fastapi/elasticsearch/stac_fastapi/elasticsearch/version.py

@@ -1,2 +1,2 @@
 """library version."""
-__version__ = "2.1.1"
+__version__ = "2.3.0"

stac_fastapi/elasticsearch/tests/clients/test_elasticsearch.py

@@ -1,3 +1,4 @@
+import time
 import uuid
 from copy import deepcopy
 from typing import Callable
@@ -256,7 +257,7 @@ def test_delete_item(
         es_core.get_item(item["id"], item["collection"], request=MockStarletteRequest)
 
 
-@pytest.mark.skip(reason="bulk not implemented")
+# @pytest.mark.skip(reason="might need a larger timeout")
 def test_bulk_item_insert(
     es_core: CoreCrudClient,
     es_transactions: TransactionsClient,
@@ -274,18 +275,18 @@ def test_bulk_item_insert(
         _item["id"] = str(uuid.uuid4())
         items.append(_item)
 
-    fc = es_core.item_collection(coll["id"], request=MockStarletteRequest)
-    assert len(fc["features"]) == 0
+    # fc = es_core.item_collection(coll["id"], request=MockStarletteRequest)
+    # assert len(fc["features"]) == 0
 
     es_bulk_transactions.bulk_item_insert(items=items)
-
+    time.sleep(3)
     fc = es_core.item_collection(coll["id"], request=MockStarletteRequest)
-    assert len(fc["features"]) == 10
+    assert len(fc["features"]) >= 10
 
-    for item in items:
-        es_transactions.delete_item(
-            item["id"], item["collection"], request=MockStarletteRequest
-        )
+    # for item in items:
+    #     es_transactions.delete_item(
+    #         item["id"], item["collection"], request=MockStarletteRequest
+    #     )
 
 
 @pytest.mark.skip(reason="Not working")