Merge pull request #1 from mgcam/project_setup

Initial project setup, ORM.

.github/workflows/run-tests.yml

@@ -0,0 +1,55 @@
+name: Test
+
+on:
+  push:
+    branches: [master, devel]
+  pull_request:
+    branches: [master, devel] 
+
+jobs:
+
+  test:
+    runs-on: ubuntu-latest
+
+    services:
+      mysql:
+        image: "mysql:8.0"
+        ports:
+          - "3306:3306"
+        options: >-
+          --health-cmd "mysqladmin ping"
+          --health-interval 10s
+          --health-timeout 5s
+          --health-retries 10
+        env:
+          MYSQL_RANDOM_ROOT_PASSWORD: yes
+          MYSQL_TCP_PORT: 3306
+          MYSQL_USER: "test"
+          MYSQL_PASSWORD: "test"
+          MYSQL_DATABASE: "study_notify"
+
+    steps:
+      - uses: actions/checkout@v4
+
+      - name: Install Poetry
+        run: |
+          pipx install poetry
+
+      - uses: actions/setup-python@v5
+        with:
+          python-version: '3.11'
+          architecture: 'x64'
+
+      - name: Run poetry install
+        run: |
+          poetry env use '3.11'
+          poetry install
+
+      - name: Run pytest
+        run: |
+          poetry run pytest
+
+      - name: Run linter (ruff)
+        run: |
+          poetry run ruff check --output-format=github .
+

README.md

@@ -1 +1,43 @@
-# npg_notifications
+# npg_notifications
+
+A utility for notifying customers about lifecycle events in the analysis
+and QC of sequencing data.
+
+[porch](https://github.com/wtsi-npg/npg_porch) service is used to queue and
+process (send) notifications. The notification producer can repeatedly send
+to `porch` the same notification. `porch` guarantees that the repeated
+notifications are not kept and therefore not processed.
+
+The consumer of the notifications is responsible for sending the message
+to the customer. For a fully automated system the consumer should implement
+the correct protocol for dealing with failed attempts to notify the customer.
+
+If different types of notification (for example, an e-mail and a MQ message)
+have to be sent for the same event, it is advised either to use a separate
+`porch` pipeline for each type of notification or to include additional
+information about the notification protocol and format into the payload that
+is sent to `porch`.
+
+## Scope
+
+The current version implements notifications for PacBio sequencing platform
+customers.
+
+## Running the scripts
+
+To register recently QC-ed entities as tasks with `porch`
+
+```bash
+qc_state_notification register --conf_file_path path/to/qc_state_app_config.ini
+```
+
+To process one `porch` task
+
+```bash
+qc_state_notification process --conf_file_path path/to/qc_state_app_config.ini
+```
+
+Processing includes claiming one task, sending per-study emails and updating the
+status of the `porch` task to `DONE`.
+
+The test data directory has an example of a [configuration file](tests/data/qc_state_app_config.ini).

pyproject.toml

@@ -0,0 +1,41 @@
+[tool.poetry]
+name = "npg_notify"
+version = "0.0.1"
+description = "Utility for client notifications"
+authors = ["Marina Gourtovaia <[email protected]>"]
+license = "GPL-3.0-or-later"
+readme = "README.md"
+
+[tool.poetry.scripts]
+qc_state_notification = "npg_notify.porch_wrapper.qc_state:run" 
+
+[tool.poetry.dependencies]
+python = "^3.11"
+SQLAlchemy = { version="^2.0.1", extras=["pymysql"] }
+SQLAlchemy-Utils = "^0.41.2"
+cryptography = { version="^41.0.3" }
+npg_porch_cli = { git="https://github.com/wtsi-npg/npg_porch_cli.git", branch="devel" }
+
+[tool.poetry.dev-dependencies]
+pytest = "^8.2.2"
+PyYAML = "^6.0.0"
+requests-mock = "^1.12.1"
+ruff = "^0.4.9"
+
+[tool.ruff]
+# Set the maximum line length to 79.
+line-length = 79
+
+[tool.ruff.lint]
+select = [
+    # flake8
+    "W",
+]
+
+[tool.pytest.ini_options]
+addopts = [
+    "--import-mode=importlib",
+]
+pythonpath = [
+  "src"
+]

src/npg_notify/config.py

@@ -0,0 +1,75 @@
+import configparser
+import json
+import pathlib
+
+"""Common utility functions for the package."""
+
+DEFAULT_CONF_FILE_TYPE = "ini"
+
+
+def get_config_data(conf_file_path: str, conf_file_section: str = None):
+    """
+    Parses a configuration file and returns its content.
+
+    Allows for two types of configuration files, 'ini' and 'json'. The type of
+    the file is determined from the extension of the file name. In case of no
+    extension an 'ini' type is assumed.
+
+    The content of the file is not cached, so subsequent calls to get data from
+    the same configuration file result in re-reading and re-parsing of the file.
+
+    Args:
+
+      conf_file_path:
+        A configuration file with database connection details.
+      conf_file_section:
+        The section of the configuration file. Optional. Should be defined
+        for 'ini' files.
+
+    Returns:
+      For an 'ini' file returns the content of the given section of the file as
+      a dictionary.
+      For a 'json' file, if the conf_file_section argument is not defined, the
+      content of a file as a Python object is returned. If the conf_file_section
+      argument is defined, the object returned by the parser is assumed to be
+      a dictionary that has the value of the 'conf_file_section' argument as a key.
+      The value corresponding to this key is returned.
+    """
+
+    conf_file_extension = pathlib.Path(conf_file_path).suffix
+    if conf_file_extension:
+        conf_file_extension = conf_file_extension[1:]
+    else:
+        conf_file_extension = DEFAULT_CONF_FILE_TYPE
+
+    if conf_file_extension == DEFAULT_CONF_FILE_TYPE:
+        if not conf_file_section:
+            raise Exception(
+                "'conf_file_section' argument is not given, "
+                "it should be defined for '{DEFAULT_CONF_FILE_TYPE}' "
+                "configuration file."
+            )
+
+        config = configparser.ConfigParser()
+        config.read(conf_file_path)
+
+        return {i[0]: i[1] for i in config[conf_file_section].items()}
+
+    elif conf_file_extension == "json":
+        conf: dict = json.load(conf_file_path)
+        if conf_file_section:
+            if isinstance(conf, dict) is False:
+                raise Exception(f"{conf_file_path} does not have sections.")
+            if conf_file_section in conf.keys:
+                conf = conf[conf_file_section]
+            else:
+                raise Exception(
+                    f"{conf_file_path} does not contain {conf_file_section} key"
+                )
+
+        return conf
+
+    else:
+        raise Exception(
+            f"Parsing for '{conf_file_extension}' files is not implemented"
+        )

src/npg_notify/db/mlwh.py

@@ -0,0 +1,148 @@
+# Copyright (C) 2024 Genome Research Ltd.
+#
+# Authors:
+#   Marina Gourtovaia <[email protected]>
+#   Kieron Taylor <[email protected]>
+#
+# This file is part of npg_notifications software package..
+#
+# npg_notifications is free software: you can redistribute it and/or modify it
+# under the terms of the GNU General Public License as published by the Free
+# Sftware Foundation; either version 3 of the License, or any later version.
+#
+# This program is distributed in the hope that it will be useful, but WITHOUT
+# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS
+# FOR A PARTICULAR PURPOSE. See the GNU General Public License for more
+# details.
+#
+# You should have received a copy of the GNU General Public License along with
+# this program. If not, see <http://www.gnu.org/licenses/>.
+
+from sqlalchemy import ForeignKey, Integer, String, UniqueConstraint, select
+from sqlalchemy.exc import NoResultFound
+from sqlalchemy.orm import (
+    DeclarativeBase,
+    Mapped,
+    Session,
+    mapped_column,
+    relationship,
+)
+
+"""
+Declarative ORM for some tables of multi-lims warehouse (mlwh) database.
+For simplicity, only columns used by this package are represented.
+
+Available ORM classes: Study, StudyUser.
+
+Utility methods: get_study_contacts.
+"""
+
+"Study contacts with these roles will receive notifications."
+ROLES = ["manager", "follower", "owner"]
+
+
+class Base(DeclarativeBase):
+    pass
+
+
+class Study(Base):
+    "A representation for the 'study' table."
+
+    __tablename__ = "study"
+
+    id_study_tmp = mapped_column(Integer, primary_key=True, autoincrement=True)
+    id_lims = mapped_column(String(10), nullable=False)
+    id_study_lims = mapped_column(String(20), nullable=False)
+    name = mapped_column(String(255))
+
+    (
+        UniqueConstraint(
+            "id_lims",
+            "id_study_lims",
+            name="study_id_lims_id_study_lims_index",
+        ),
+    )
+
+    study_users: Mapped[set["StudyUser"]] = relationship()
+
+    def __repr__(self):
+        return f"Study {self.id_study_lims}, {self.name}"
+
+    def contacts(self) -> list[str]:
+        """Retrieves emails of contacts for this study object.
+
+        Returns:
+          A sorted list of unique emails for managers, followers or owners of
+          the study.
+        """
+
+        # In order to eliminate repetition, the comprehension expression below
+        # returns a set, which is then sorted to return a sorted list.
+        return sorted(
+            {
+                u.email
+                for u in self.study_users
+                if (u.email is not None and u.role is not None)
+                and (u.role in ROLES)
+            }
+        )
+
+
+class StudyUser(Base):
+    "A representation for the 'study_users' table."
+
+    __tablename__ = "study_users"
+
+    id_study_users_tmp = mapped_column(
+        Integer, primary_key=True, autoincrement=True
+    )
+    id_study_tmp = mapped_column(
+        Integer, ForeignKey("study.id_study_tmp"), nullable=False, index=True
+    )
+    role = mapped_column(String(255), nullable=True)
+    email = mapped_column(String(255), nullable=True)
+
+    study: Mapped["Study"] = relationship(back_populates="study_users")
+
+    def __repr__(self):
+        role = self.role if self.role else "None"
+        email = self.email if self.email else "None"
+        return f"StudyUser role={role}, email={email}"
+
+
+class StudyNotFoundError(Exception):
+    "An error to use when a study does not exist in mlwh."
+
+    pass
+
+
+def get_study_contacts(session: Session, id: str) -> list[str]:
+    """Retrieves emails of study contacts from the mlwh database.
+
+    Args:
+      session:
+        sqlalchemy.orm.Session object
+      id:
+        String study ID.
+
+    Returns:
+      A sorted list of unique emails for managers, followers or owners of
+      the study.
+
+    Example:
+
+      from npg_notify.db.mlwh get_study_contacts
+      contact_emails = get_study_contacts(session=session, id="5901")
+    """
+    try:
+        contacts = (
+            session.execute(select(Study).where(Study.id_study_lims == id))
+            .scalar_one()
+            .contacts()
+        )
+    except NoResultFound:
+        raise StudyNotFoundError(
+            f"Study with ID {id} is not found in ml warehouse"
+        )
+
+    return contacts