add manager methods for creating from GH API data (#9)

* add `Installation` manager method for creating from GH API data

* add methods to Repo too

* update readme and changelog

* add tests for repo methods

* add success codes so coverage report will print fully

* add single repo data test case

CHANGELOG.md

@@ -18,6 +18,10 @@ and this project attempts to adhere to [Semantic Versioning](https://semver.org/
 
 ## [Unreleased]
 
+### Added
+
+- Added `acreate_from_gh_data`/`create_from_gh_data` manager methods to `Installation` and `Repository` models.
+
 ## [0.1.0]
 
 ### Added

README.md

@@ -249,6 +249,7 @@ async with AsyncGitHubAPI(installation_id=installation.installation_id) as gh:
 ##### Manager methods
 
 - `acreate_from_event`/`create_from_event`: Create from installation events _(primarily for internal use)_
+- `acreate_from_gh_data`/`create_from_gh_data`: Create from GitHub API response data _(primarily for internal use)_
 - `aget_from_event`/`get_from_event`: Retrieve installation from webhook events (`gidgethub.sansio.Event`)
 
 ##### Model methods
@@ -269,6 +270,7 @@ issues = await repo.aget_issues(params={"state": "open"})
 
 ##### Manager methods
 
+- `acreate_from_gh_data`/`create_from_gh_data`: Create from GitHub API response data _(primarily for internal use)_
 - `aget_from_event`/`get_from_event`: Retrieve repository from webhook events (`gidgethub.sansio.Event`)
 
 ##### Model methods

noxfile.py

@@ -108,8 +108,12 @@ def coverage(session):
     try:
         session.run("python", "-m", "pytest", "--cov", "--cov-report=")
     finally:
+        # 0 -> OK
+        # 2 -> code coverage percent unmet
+        success_codes = [0, 2]
+
         report_cmd = ["python", "-m", "coverage", "report"]
-        session.run(*report_cmd)
+        session.run(*report_cmd, success_codes=success_codes)
 
         if summary := os.getenv("GITHUB_STEP_SUMMARY"):
             report_cmd.extend(["--skip-covered", "--skip-empty", "--format=markdown"])
@@ -118,10 +122,18 @@ def coverage(session):
                 output_buffer.write("")
                 output_buffer.write("### Coverage\n\n")
                 output_buffer.flush()
-                session.run(*report_cmd, stdout=output_buffer)
+                session.run(
+                    *report_cmd, stdout=output_buffer, success_codes=success_codes
+                )
         else:
             session.run(
-                "python", "-m", "coverage", "html", "--skip-covered", "--skip-empty"
+                "python",
+                "-m",
+                "coverage",
+                "html",
+                "--skip-covered",
+                "--skip-empty",
+                success_codes=success_codes,
             )
 
 

src/django_github_app/models.py

@@ -64,34 +64,26 @@ def action(self) -> str | None:
 
 class InstallationManager(models.Manager["Installation"]):
     async def acreate_from_event(self, event: sansio.Event):
-        installation_data = event.data["installation"]
-
-        app_id = installation_data["app_id"]
+        app_id = event.data["installation"]["app_id"]
 
         if str(app_id) == app_settings.APP_ID:
-            installation = await self.acreate(
-                installation_id=installation_data["id"],
-                data=installation_data,
-            )
+            installation = await self.acreate_from_gh_data(event.data["installation"])
 
-            repository_data = event.data["repositories"]
-
-            repositories = [
-                Repository(
-                    installation=installation,
-                    repository_id=repository["id"],
-                    repository_node_id=repository["node_id"],
-                    full_name=repository["full_name"],
-                )
-                for repository in repository_data
-            ]
-            await Repository.objects.abulk_create(repositories)
+            await Repository.objects.acreate_from_gh_data(
+                event.data["repositories"], installation
+            )
 
             return installation
 
     def create_from_event(self, event: sansio.Event):
         return async_to_sync(self.acreate_from_event)(event)
 
+    async def acreate_from_gh_data(self, data: dict[str, str]):
+        return await self.acreate(installation_id=data["id"], data=data)
+
+    def create_from_gh_data(self, data: dict[str, str]):
+        return async_to_sync(self.acreate_from_gh_data)(data)
+
     async def aget_from_event(self, event: sansio.Event):
         try:
             installation_id = event.data["installation"]["id"]
@@ -147,6 +139,33 @@ def get_access_token(self, gh: abc.GitHubAPI):  # pragma: no cover
 
 
 class RepositoryManager(models.Manager["Repository"]):
+    async def acreate_from_gh_data(
+        self, data: dict[str, str] | list[dict[str, str]], installation: Installation
+    ):
+        if isinstance(data, list):
+            repositories = [
+                Repository(
+                    installation=installation,
+                    repository_id=repository["id"],
+                    repository_node_id=repository["node_id"],
+                    full_name=repository["full_name"],
+                )
+                for repository in data
+            ]
+            return await Repository.objects.abulk_create(repositories)
+        else:
+            return await self.acreate(
+                installation=installation,
+                repository_id=data["id"],
+                repository_node_id=data["node_id"],
+                full_name=data["full_name"],
+            )
+
+    def create_from_gh_data(
+        self, data: dict[str, str] | list[dict[str, str]], installation: Installation
+    ):
+        return async_to_sync(self.acreate_from_gh_data)(data, installation)
+
     async def aget_from_event(self, event: sansio.Event):
         try:
             repository_id = event.data["repository"]["id"]

tests/test_models.py

@@ -160,6 +160,31 @@ def test_create_from_event(self, create_event, override_app_settings):
             repositories
         )
 
+    @pytest.mark.asyncio
+    async def test_acreate_from_gh_data(self):
+        installation_data = {
+            "id": seq.next(),
+            "app_id": seq.next(),
+        }
+
+        installation = await Installation.objects.acreate_from_gh_data(
+            installation_data
+        )
+
+        assert installation.installation_id == installation_data["id"]
+        assert installation.data == installation_data
+
+    def test_create_from_gh_data(self):
+        installation_data = {
+            "id": seq.next(),
+            "app_id": seq.next(),
+        }
+
+        installation = Installation.objects.create_from_gh_data(installation_data)
+
+        assert installation.installation_id == installation_data["id"]
+        assert installation.data == installation_data
+
     @pytest.mark.asyncio
     async def test_aget_from_event(self, ainstallation, create_event):
         installation = await ainstallation
@@ -213,6 +238,60 @@ def test_from_event_invalid_action(self, create_event):
 
 
 class TestRepositoryManager:
+    @pytest.mark.asyncio
+    async def test_acreate_from_gh_data_list(self, ainstallation):
+        installation = await ainstallation
+        data = [
+            {"id": seq.next(), "node_id": "node1", "full_name": "owner/repo1"},
+            {"id": seq.next(), "node_id": "node2", "full_name": "owner/repo2"},
+        ]
+
+        repositories = await Repository.objects.acreate_from_gh_data(data, installation)
+
+        assert len(repositories) == len(data)
+        for i, repo in enumerate(repositories):
+            assert repo.repository_id == data[i]["id"]
+            assert repo.repository_node_id == data[i]["node_id"]
+            assert repo.full_name == data[i]["full_name"]
+            assert repo.installation_id == installation.id
+
+    def test_create_from_gh_data_list(self, installation):
+        data = [
+            {"id": seq.next(), "node_id": "node1", "full_name": "owner/repo1"},
+            {"id": seq.next(), "node_id": "node2", "full_name": "owner/repo2"},
+        ]
+
+        repositories = Repository.objects.create_from_gh_data(data, installation)
+
+        assert len(repositories) == len(data)
+        for i, repo in enumerate(repositories):
+            assert repo.repository_id == data[i]["id"]
+            assert repo.repository_node_id == data[i]["node_id"]
+            assert repo.full_name == data[i]["full_name"]
+            assert repo.installation_id == installation.id
+
+    @pytest.mark.asyncio
+    async def test_acreate_from_gh_data_single(self, ainstallation):
+        installation = await ainstallation
+        data = {"id": seq.next(), "node_id": "node1", "full_name": "owner/repo1"}
+
+        repository = await Repository.objects.acreate_from_gh_data(data, installation)
+
+        assert repository.repository_id == data["id"]
+        assert repository.repository_node_id == data["node_id"]
+        assert repository.full_name == data["full_name"]
+        assert repository.installation_id == installation.id
+
+    def test_create_from_gh_data_single(self, installation):
+        data = {"id": seq.next(), "node_id": "node1", "full_name": "owner/repo1"}
+
+        repository = Repository.objects.create_from_gh_data(data, installation)
+
+        assert repository.repository_id == data["id"]
+        assert repository.repository_node_id == data["node_id"]
+        assert repository.full_name == data["full_name"]
+        assert repository.installation_id == installation.id
+
     @pytest.mark.asyncio
     async def test_aget_from_event(self, arepository, create_event):
         repository = await arepository