Upgrade Elasticsearch from 8.6.1 to 8.17.0 (#18)

.env.sample

@@ -21,7 +21,7 @@ ELASTICSEARCH_VERIFY_CERTS=False
 ELASTICSEARCH_TIMEOUT=600
 
 # Version of Elastic products
-STACK_VERSION=8.6.1
+STACK_VERSION=8.17.0
 
 # Set the cluster name
 CLUSTER_NAME=docker-cluster

.env.test

@@ -17,4 +17,4 @@ ELASTICSEARCH_SCHEME=http
 ELASTICSEARCH_TIMEOUT=600
 
 # Version of Elastic products
-STACK_VERSION=8.6.1
+STACK_VERSION=8.17.0

.github/workflows/test.yml

@@ -22,7 +22,7 @@ jobs:
       - name: Run image
         uses: abatilo/actions-poetry@v2
         with:
-          poetry-version: 1.4.0
+          poetry-version: 1.8.4
       - name: Install dependencies
         run: poetry install
       - name: Black
@@ -34,7 +34,7 @@ jobs:
       - name: Copy test envvars
         run: cp .env.test .env
       - name: Start docker containers
-        run: docker-compose up --detach
+        run: docker compose up --detach
       - name: Tests
         run: poetry run pytest --cov=src --cov-report=xml
       - name: Upload coverage reports to Codecov

README.md

@@ -307,9 +307,9 @@ and [`docker-compose`](https://docs.docker.com/compose/install/) installed.
 
 ```shell
 % docker-compose --version
-Docker Compose version v2.15.1
+Docker Compose version v2.31.0
 % poetry --version
-Poetry (version 1.4.0)
+Poetry (version 1.8.4)
 ```
 
 Clone the repo:
@@ -334,6 +334,7 @@ docker-compose --env-file=.env.test up -d
 Run tests (with -s option to allow more verbose output):
 
 ```shell
+cp .env.test .env # copy the test environment variables
 poetry run pytest -s
 ```
 

docker-compose.yml

@@ -1,5 +1,3 @@
-version: '3.9'
-
 services:
   postgres:
     image: postgres:15

pyproject.toml

@@ -11,8 +11,8 @@ sinker = "sinker.__main__:main"
 
 [tool.poetry.dependencies]
 python = "^3.9"
-elasticsearch = "^8.6.1"
-environs = ">=9.5,<12.0"
+elasticsearch = "^8.17.0"
+environs = ">=9.5,<15.0"
 psycopg = "^3.1.8"
 pytest-mock = "^3.10.0"
 

src/sinker/runner.py

@@ -111,5 +111,7 @@ def iterate(self):
         # activity, like inserts into a schema/table you aren't synchronizing to Elasticsearch. If you are worried
         # about your replication slot growing too large during a scenario like this, you can periodically trigger
         # a materialized view refresh to clear out the slot (see the CREATE_TODO_ENTRY query template).
-        processed_tuples, _ = bulk(get_client(), self.bulk_gen.generate_actions(), **ELASTICSEARCH_BULK_KWARGS)
+        processed_tuples, _ = bulk(
+            client=get_client(), actions=self.bulk_gen.generate_actions(), stats_only=False, **ELASTICSEARCH_BULK_KWARGS
+        )
         logger.info(f"Processed {processed_tuples} tuples from replication slot")

src/sinker/settings.py

@@ -13,7 +13,7 @@
 
 # Sinker:
 # path to the views/indices configs
-SINKER_DEFINITIONS_PATH = env.str("SINKER_DEFINITIONS_PATH", default=None)
+SINKER_DEFINITIONS_PATH = env.str("SINKER_DEFINITIONS_PATH", default=".")
 SINKER_SCHEMA = env.str("SINKER_SCHEMA", default=DEFAULT_SCHEMA)
 SINKER_REPLICATION_SLOT = env.str("SINKER_REPLICATION_SLOT", default="sinker")
 SINKER_TODO_TABLE = env.str("SINKER_TODO_TABLE", default="todo")

src/sinker/sinker.py

@@ -52,7 +52,9 @@ def setup_es(self) -> None:
     def backfill_index(self) -> None:
         # populate ES index with initial data from materialized view
         logger.info(f"Populating {self.index} with initial data from {self.view}")
-        added_docs, _ = bulk(get_client(), self.backfill_stream(), **ELASTICSEARCH_BULK_KWARGS)
+        added_docs, _ = bulk(
+            client=get_client(), actions=self.backfill_stream(), stats_only=False, **ELASTICSEARCH_BULK_KWARGS
+        )
         logger.info(f"Added {added_docs} documents to {self.index}")
 
     def backfill_stream(self) -> Iterable[Dict[str, Any]]: