Extract ConfigLoaders, and add Github config loader for raw user content

Author: DavidSouther

aws_doc_sdk_examples_tools/doc_gen.py

@@ -3,7 +3,9 @@
 
 import yaml
 import json
+import requests
 
+from abc import ABC, abstractmethod
 from collections import defaultdict
 from dataclasses import dataclass, field, fields, is_dataclass, asdict
 from functools import reduce
@@ -47,6 +49,46 @@ class DocGenMergeWarning(MetadataError):
     pass
 
 
+class ConfigLoader(ABC):
+    @abstractmethod
+    def load(self, filename: str) -> Tuple[Path, Any]:
+        pass
+
+
+class NoneLoader(ConfigLoader):
+    def load(self, filename: str) -> Tuple[Path, Any]:
+        return Path(filename), yaml.safe_load("")
+
+
+class FileLoader(ConfigLoader):
+    def __init__(self, root: Optional[Path] = None):
+        self.config = root or Path(__file__).parent / "config"
+
+    def load(self, filename: str) -> Tuple[Path, Any]:
+        path = self.config / filename
+        print(f"Opening {path}")
+        with path.open(encoding="utf-8") as file:
+            return path, yaml.safe_load(file)
+
+
+class GitHubLoader(ConfigLoader):
+    def __init__(self, repo: Optional[str] = None, commit: Optional[str] = None):
+        self.repo = repo or "awsdocs/aws-doc-sdk-examples-tools"
+        self.commit = (
+            commit or "refs/heads/main"
+        )  # or refs/tags/2025.07.0 or a specific SHA
+        self.path = f"{self.repo}/{self.commit}/aws_doc_sdk_examples_tools/config"
+
+    def load(self, filename: str) -> Tuple[Path, Any]:
+        path = f"{self.path}/{filename}"
+        url = f"https://raw.githubusercontent.com/{path}"
+        print(f"Requesting {url}")
+        r = requests.get(url)
+        if r.status_code == 200:
+            return Path(path), yaml.safe_load(r.text)
+        raise Exception(f"Failed to request {url} ({r.status_code} {r.text})")
+
+
 @dataclass
 class DocGen:
     root: Path
@@ -170,7 +212,9 @@ def empty(cls, validation: ValidationConfig = ValidationConfig()) -> "DocGen":
 
     @classmethod
     def default(cls) -> "DocGen":
-        return DocGen.empty().for_root(Path(__file__).parent, incremental=True)
+        return DocGen.empty().for_root(
+            Path(__file__).parent, GitHubLoader(), incremental=True
+        )
 
     def clone(self) -> "DocGen":
         return DocGen(
@@ -186,13 +230,9 @@ def clone(self) -> "DocGen":
             examples={},
         )
 
-    def for_root(
-        self, root: Path, config: Optional[Path] = None, incremental=False
-    ) -> "DocGen":
+    def for_root(self, root: Path, loader: ConfigLoader, incremental=False) -> "DocGen":
         self.root = root
 
-        config = config or Path(__file__).parent / "config"
-
         try:
             with open(root / ".doc_gen" / "validation.yaml", encoding="utf-8") as file:
                 validation = yaml.safe_load(file)
@@ -203,46 +243,37 @@ def for_root(
             pass
 
         try:
-            sdk_path = config / "sdks.yaml"
-            with sdk_path.open(encoding="utf-8") as file:
-                meta = yaml.safe_load(file)
-                sdks, errs = parse_sdks(sdk_path, meta)
-                self.sdks = sdks
-                self.errors.extend(errs)
+            sdk_path, meta = loader.load("sdks.yaml")
+            sdks, errs = parse_sdks(sdk_path, meta)
+            self.sdks = sdks
+            self.errors.extend(errs)
         except Exception:
             pass
 
         try:
-            services_path = config / "services.yaml"
-            with services_path.open(encoding="utf-8") as file:
-                meta = yaml.safe_load(file)
-                services, service_errors = parse_services(services_path, meta)
-                self.services = services
-                for service in self.services.values():
-                    if service.expanded:
-                        self.entities[service.long] = service.expanded.long
-                        self.entities[service.short] = service.expanded.short
-                self.errors.extend(service_errors)
+            services_path, meta = loader.load("services.yaml")
+            services, service_errors = parse_services(services_path, meta)
+
+            self.services = services
+            for service in self.services.values():
+                if service.expanded:
+                    self.entities[service.long] = service.expanded.long
+                    self.entities[service.short] = service.expanded.short
+            self.errors.extend(service_errors)
         except Exception:
             pass
 
         try:
-            categories_path = config / "categories.yaml"
-            with categories_path.open(encoding="utf-8") as file:
-                meta = yaml.safe_load(file)
-                standard_categories, categories, errs = parse_categories(
-                    categories_path, meta
-                )
-                self.standard_categories = standard_categories
-                self.categories = categories
-                self.errors.extend(errs)
+            path, meta = loader.load("categories.yaml")
+            standard_categories, categories, errs = parse_categories(path, meta)
+            self.standard_categories = standard_categories
+            self.categories = categories
+            self.errors.extend(errs)
         except Exception:
             pass
 
         try:
-            entities_config_path = config / "entities.yaml"
-            with entities_config_path.open(encoding="utf-8") as file:
-                entities_config = yaml.safe_load(file)
+            path, entities_config = loader.load("entities.yaml")
             for entity, expanded in entities_config["expanded_override"].items():
                 self.entities[entity] = expanded
         except Exception:
@@ -294,12 +325,13 @@ def process_metadata(self, path: Path) -> "DocGen":
     def from_root(
         cls,
         root: Path,
-        config: Optional[Path] = None,
+        loader: Optional[ConfigLoader] = None,
         validation: ValidationConfig = ValidationConfig(),
         incremental: bool = False,
     ) -> "DocGen":
+        loader = loader or GitHubLoader()
         return DocGen.empty(validation=validation).for_root(
-            root, config, incremental=incremental
+            root, loader, incremental=incremental
         )
 
     def validate(self):

aws_doc_sdk_examples_tools/doc_gen_test.py

@@ -10,7 +10,7 @@
 import json
 
 from .categories import Category, TitleInfo
-from .doc_gen import DocGen, DocGenEncoder
+from .doc_gen import DocGen, DocGenEncoder, FileLoader
 from .metadata import Example
 from .metadata_errors import MetadataErrors, MetadataError
 from .sdks import Sdk, SdkVersion
@@ -65,7 +65,7 @@ def test_merge(a: DocGen, b: DocGen, d: DocGen):
 def test_incremental():
     errors = MetadataErrors()
     doc_gen = DocGen(Path(), errors).for_root(
-        Path(__file__).parent / "test_resources", incremental=False
+        Path(__file__).parent / "test_resources", loader=FileLoader(), incremental=False
     )
     assert len(doc_gen.examples) == 0
     doc_gen.process_metadata(doc_gen.root / "awsentity_metadata.yaml")
@@ -231,7 +231,7 @@ def test_doc_gen_encoder(sample_doc_gen: DocGen):
 def test_doc_gen_load_snippets():
     errors = MetadataErrors()
     doc_gen = DocGen(Path(), errors).for_root(
-        Path(__file__).parent / "test_resources", incremental=False
+        Path(__file__).parent / "test_resources", loader=FileLoader(), incremental=False
     )
     doc_gen.process_metadata(doc_gen.root / "valid_metadata.yaml")
     doc_gen.collect_snippets()

aws_doc_sdk_examples_tools/sdks.py

@@ -182,6 +182,7 @@ def parse(file: Path, yaml: Dict[str, Any]) -> tuple[Dict[str, Sdk], MetadataErr
     from pathlib import Path
 
     path = Path(__file__).parent.parent.parent / ".doc_gen" / "metadata" / "sdks.yaml"
+
     with open(path) as file:
         meta = yaml.safe_load(file)
     examples, errors = parse(path, meta)

aws_doc_sdk_examples_tools/stats.py

@@ -2,14 +2,16 @@
 from pathlib import Path
 from pprint import pformat
 
-from .doc_gen import DocGen
+from .doc_gen import DocGen, FileLoader
 
 
 def main(roots: List[str]):
     base = DocGen.empty()
     for root in roots:
         docgen_root = Path(root)
-        doc_gen = base.clone().for_root(docgen_root)
+        doc_gen = base.clone().for_root(
+            docgen_root, loader=FileLoader(Path(root) / ".doc_gen" / "config")
+        )
         doc_gen.collect_snippets()
         print(f"Root	{docgen_root.name}")
         stats = doc_gen.stats()

aws_doc_sdk_examples_tools/validate.py

@@ -6,7 +6,7 @@
 from pathlib import Path
 from sys import exit
 
-from .doc_gen import DocGen
+from .doc_gen import DocGen, FileLoader
 from .project_validator import check_files, verify_sample_files, ValidationConfig
 
 
@@ -32,11 +32,22 @@ def main():
         "must have them.",
         required=False,
     )
+    parser.add_argument(
+        "--local_loader",
+        action="store_true",
+        default=False,
+        help="Use LocalConfigLoader, instead of GithubConfigLoader",
+        required=False,
+    )
     args = parser.parse_args()
     root_path = Path(args.root).resolve()
 
+    loader = FileLoader() if args.local_loader else None
+
     doc_gen = DocGen.from_root(
-        root=root_path, validation=ValidationConfig(strict_titles=args.strict_titles)
+        root=root_path,
+        loader=loader,
+        validation=ValidationConfig(strict_titles=args.strict_titles),
     )
     doc_gen.collect_snippets(snippets_root=root_path)
     doc_gen.validate()