Feat/metrix proxy

README.md

@@ -57,7 +57,7 @@ Below is an example of a **Python-defined pipeline** that mirrors what most team
 build, lint, test, coverage, and deploy — all orchestrated through `pygha`.
 
 ```python
-from pygha import job, default_pipeline
+from pygha import job, default_pipeline, matrix
 from pygha.steps import run, checkout, uses
 
 # Configure the default pipeline to run on main push and PRs
@@ -72,7 +72,7 @@ def test_matrix():
     checkout()
 
     # Use the matrix variable in your step arguments
-    setup_python("${{ matrix.python }}", cache="pip")
+    setup_python(matrix.python, cache="pip")
 
     run("pip install .[dev]")
     run("pytest")

changelog.md

@@ -5,6 +5,12 @@ All notable changes to this project will be documented in this file.
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
+## [0.4.0] - 2025-12-22
+### Added
+- **Matrix Support**: Added matrix proxy object and build-time validation to prevent usage of undefined matrix variables. ([#62])
+
+[#62]: https://github.com/parneetsingh022/pygha/issues/62
+
 ## [0.3.1] - 2025-12-22
 
 ### Fixed

docs/overview.rst

@@ -132,13 +132,13 @@ In this example, the ``test`` job runs three times, once for each Python version
 
 .. code-block:: python
 
-   from pygha import job
-   from pygha.steps import run
+   from pygha import job, matrix
+   from pygha.steps import echo
 
    @job(matrix={"python": ["3.11", "3.12", "3.13"]})
    def test():
        # Access the matrix context in your shell commands
-       run("echo Running tests on Python ${{ matrix.python }}")
+       echo(f"Running tests on Python { matrix.python }")
 
 Dynamic Runners (OS Matrix)
 ~~~~~~~~~~~~~~~~~~~~~~~~~~~
@@ -150,11 +150,11 @@ for cross-platform testing.
 
    @job(
        name="build",
-       runs_on="${{ matrix.os }}",
+       runs_on=matrix.os,
        matrix={"os": ["ubuntu-latest", "macos-latest", "windows-latest"]}
    )
    def build_os():
-       run("echo Building on ${{ matrix.os }}")
+       run(f"echo Building on { matrix.os }")
 
 Fail Fast
 ~~~~~~~~~
@@ -169,7 +169,7 @@ To let all jobs finish regardless of failure, set ``fail_fast=False``.
        fail_fast=False
    )
    def long_running_test():
-       run("./run_tests.sh --shard ${{ matrix.shard }}")
+       run(f"./run_tests.sh --shard { matrix.shard }")
 
 Job Timeout
 ~~~~~~~~~~~

pyproject.toml

@@ -4,7 +4,8 @@ build-backend = "hatchling.build"
 
 [project]
 name = "pygha"
-version = "0.3.1"
+version = "0.4.0"
+
 description = "A Python-native CI/CD framework for defining, testing, and transpiling pipelines."
 readme = "README.md"
 requires-python = ">=3.11"

recipe/meta.yaml

@@ -1,6 +1,8 @@
 package:
   name: pygha
-  version: "0.3.1"
+
+  version: "0.4.0"
+
 
 source:
   path: ..

src/pygha/__init__.py

@@ -1,6 +1,8 @@
 # pygha/__init__.py
 from .decorators import job
 from pygha.registry import pipeline, default_pipeline
+from pygha.expr import matrix
+
+__version__ = "0.4.0"
+__all__ = ["job", "pipeline", "default_pipeline", "matrix"]
 
-__version__ = "0.3.1"
-__all__ = ["job", "pipeline", "default_pipeline"]

src/pygha/expr.py

@@ -55,3 +55,23 @@ def success() -> Expression:
 
 def failure() -> Expression:
     return Expression("failure()")
+
+
+class MatrixProxy:
+    def __getattr__(self, name: str) -> str:
+        # Dot notation: "Magic" convenience for standard GHA keys.
+        # Python forces us to use underscores (matrix.python_version),
+        # so we automatically convert them to hyphens for GHA.
+        # matrix.python_version -> ${{ matrix.python-version }}
+        gha_name = name.replace("_", "-")
+        return f"${{{{ matrix.{gha_name} }}}}"
+
+    def __getitem__(self, name: str) -> str:
+        # Bracket notation: "Strict" mode.
+        # Used when the user specifically needs an underscore or special char.
+        # matrix['python_version'] -> ${{ matrix.python_version }}
+        # matrix['python-version'] -> ${{ matrix.python-version }}
+        return f"${{{{ matrix.{name} }}}}"
+
+
+matrix = MatrixProxy()

src/pygha/transpilers/github.py

@@ -1,3 +1,4 @@
+import re
 from ruamel.yaml import YAML
 from ruamel.yaml.comments import CommentedMap
 
@@ -6,7 +7,7 @@
 from collections.abc import MutableMapping
 
 from collections.abc import Iterable
-from ..models import Pipeline
+from ..models import Pipeline, Job
 from ..registry import get_default
 
 
@@ -19,6 +20,52 @@ def _sorted_unique(items: Iterable[str]) -> list[str]:
         # Ensure deterministic, duplicate-free 'needs'
         return sorted(set(items))
 
+    def _extract_vars(self, text: str) -> set[str]:
+        """Extracts 'matrix.xxx' from a string like '${{ matrix.xxx }}'."""
+        # Matches ${{ matrix.os }}, ${{ matrix.python-version }}, etc.
+        return set(re.findall(r"\$\{\{\s*matrix\.([\w-]+)\s*\}\}", text))
+
+    def _scan_for_vars(self, data: Any) -> set[str]:
+        """Recursively scans a dict, list, or string for matrix variables."""
+        found = set()
+        if isinstance(data, str):
+            found.update(self._extract_vars(data))
+        elif isinstance(data, dict):
+            for value in data.values():
+                found.update(self._scan_for_vars(value))
+        elif isinstance(data, list):
+            for item in data:
+                found.update(self._scan_for_vars(item))
+        return found
+
+    def _validate_matrix(self, job: Job, job_dict: dict[str, Any]) -> None:
+        """Ensures all matrix variables used in the job are actually defined."""
+        used_vars = self._scan_for_vars(job_dict)
+
+        if not used_vars:
+            return
+
+        if not job.matrix:
+            # Get the first invalid variable found for the error message
+            invalid = next(iter(used_vars))
+            raise ValueError(
+                f"Job '{job.name}' uses '${{{{ matrix.{invalid} }}}}' but has no matrix defined."
+            )
+
+        valid_keys = {k for k in job.matrix.keys() if k not in ("include", "exclude")}
+
+        if "include" in job.matrix:
+            for item in job.matrix["include"]:
+                if isinstance(item, dict):
+                    valid_keys.update(item.keys())
+
+        unknowns = used_vars - valid_keys
+        if unknowns:
+            raise ValueError(
+                f"Job '{job.name}' uses undefined matrix variables: {sorted(unknowns)}. "
+                f"Available keys: {sorted(valid_keys)}"
+            )
+
     def to_dict(self) -> MutableMapping[str, Any]:
         jobs_dict: dict[str, Any] = {}
 
@@ -36,7 +83,6 @@ def to_dict(self) -> MutableMapping[str, Any]:
             if job.matrix:
                 strategy: dict[str, Any] = {"matrix": job.matrix}
 
-                # Only add fail-fast if the user explicitly set it (True or False)
                 if job.fail_fast is not None:
                     strategy["fail-fast"] = job.fail_fast
 
@@ -56,9 +102,10 @@ def to_dict(self) -> MutableMapping[str, Any]:
                     d["if"] = step.if_condition
                 steps_list.append(d)
 
-            # Now add steps
             job_dict["steps"] = steps_list
 
+            self._validate_matrix(job, job_dict)
+
             jobs_dict[job.name] = job_dict
 
         workflow: MutableMapping[str, Any] = CommentedMap()