Merge pull request #3 from lsst-dm/tickets/DM-37511

DM-37511: Summarize Header Service into ConsDB.

.github/workflows/build.yaml

@@ -0,0 +1,26 @@
+name: CI build of all containers
+on:
+  push:
+    branches:
+      - main
+    tags:
+      - "*"
+  pull_request:
+
+jobs:
+  push:
+    runs-on: ubuntu-latest
+    permissions:
+      packages: write
+      contents: read
+
+    steps:
+    - name: Checkout
+      uses: actions/checkout@v4
+
+    - name: Build hinfo
+      uses: lsst-sqre/build-and-push-to-ghcr@v1
+      with:
+        image: ${{ github.repository }}-hinfo
+        github_token: ${{ secrets.GITHUB_TOKEN }}
+        dockerfile: Dockerfile.hinfo

.github/workflows/lint.yaml

@@ -8,12 +8,12 @@ jobs:
   lint:
     runs-on: ubuntu-latest
     steps:
-      - uses: actions/checkout@v2
+      - uses: actions/checkout@v4
 
       - name: Set up Python
-        uses: actions/setup-python@v2
+        uses: actions/setup-python@v5
         with:
-          python-version: 3.9
+          python-version: '3.11'
 
       - name: Install
         run: pip install -r <(curl https://raw.githubusercontent.com/lsst/linting/main/requirements.txt)

Dockerfile.hinfo

@@ -0,0 +1,22 @@
+ARG RUBINENV_VERSION=8.0.0
+FROM lsstsqre/newinstall:${RUBINENV_VERSION}
+ARG OBS_LSST_VERSION
+ENV OBS_LSST_VERSION=${OBS_LSST_VERSION:-w_2024_06}
+USER lsst
+RUN source loadLSST.bash && mamba install aiokafka httpx
+RUN source loadLSST.bash && pip install kafkit
+RUN source loadLSST.bash && eups distrib install -t "${OBS_LSST_VERSION}" obs_lsst
+COPY python/lsst/consdb/hinfo.py ./hinfo/
+
+# Environment variables that must be set:
+#   INSTRUMENT: LATISS, LSSTComCam, LSSTComCamSim, LSSTCam
+#   POSTGRES_URL: SQLAlchemy connection URL
+#   KAFKA_BOOTSTRAP: host:port of bootstrap server
+#   KAFKA_PASSWORD: password for SASL_PLAIN authentication
+#   SCHEMA_URL: Kafkit registry schema URL
+# Optional environment variables:
+#   BUCKET_PREFIX: set to "rubin:" at USDF, default is ""
+#   KAFKA_GROUP_ID: name of consumer group, default is "consdb-consumer"
+#   KAFKA_USERNAME: username for SASL_PLAIN authentication, default is "consdb"
+
+ENTRYPOINT [ "bash", "-c", "source loadLSST.bash; setup obs_lsst; python ./hinfo/hinfo.py" ]

Dockerfile.server

@@ -0,0 +1,8 @@
+FROM python:3.11
+RUN pip install flask gunicorn sqlalchemy
+WORKDIR /consdb-server
+COPY src/server.py /consdb-server/
+# Environment variables that must be set:
+# POSTGRES_URL
+ENTRYPOINT [ "gunicorn", "-b", "0.0.0.0:8000", "-w", "2", "server:app" ]
+

README.rst

@@ -2,4 +2,6 @@
 consdb
 ######
 
-This CSC listens for SAL events, executes EFD queries when they arrive, and writes results to columns in relational database tables in the Consolidated Database.
+Scripts and services for generating the Summit Visit Database and Consolidated Database (ConsDB), including summarizing the Engineering and Facilities Database (EFD).
+
+See also DMTN-227.lsst.io

python/lsst/consdb/__init__.py

@@ -19,7 +19,4 @@
 # You should have received a copy of the GNU General Public License
 # along with this program.  If not, see <https://www.gnu.org/licenses/>.
 
-try:
-    from .version import *  # Generated by sconsUtils
-except ImportError:
-    __version__ = "?"
+# from .version import *  # Generated by sconsUtils

python/lsst/consdb/client.py

@@ -0,0 +1,58 @@
+import os
+from pandas import DataFrame
+import requests
+from requests.exceptions import RequestException
+from typing import Any, Iterable
+from urllib.parse import urljoin
+
+session = requests.Session()
+base_url = os.environ["CONSDB_URL"]
+
+
+def insert(table: str, values: dict[str, Any], **kwargs):
+    values.update(kwargs)
+    # check values against schema for table
+    data = {"table": table, "values": values}
+    url = urljoin(base_url, "insert")
+    try:
+        response = requests.post(url, json=data)
+    except RequestException as e:
+        raise e
+    response.raise_for_status()
+
+
+def query(
+    tables: str | Iterable[str],
+    columns: str | Iterable[str],
+    *,
+    where: str | None = None,
+    join: str | None = None
+) -> list[Any]:
+    if isinstance(tables, str):
+        tables = [tables]
+    if isinstance(columns, str):
+        columns = [columns]
+    url = urljoin(base_url, "query")
+    data = {"tables": tables, "columns": columns, "where": where, "join": join}
+    try:
+        response = requests.post(url, json=data)
+    except RequestException as e:
+        raise e
+    try:
+        response.raise_for_status()
+    except Exception as ex:
+        print(response.content.decode())
+        raise ex
+    arr = response.json()
+    return DataFrame(arr[1:], columns=arr[0])
+
+
+def schema(table: str):
+    url = urljoin(base_url, "schema/")
+    url = urljoin(url, table)
+    try:
+        response = requests.get(url)
+    except RequestException as e:
+        raise e
+    response.raise_for_status()
+    return response.json()