feat: add created_at option (#59)

depwatch/command.py

@@ -1,4 +1,5 @@
 from dotenv import load_dotenv
+from depwatch.date_utils import DateRange
 
 from depwatch.history import (
     convert_repository_history_to_workflow_ids,
@@ -10,13 +11,17 @@
 
 
 def generate_histories(
-    name: str, code_only: bool, limit: int, workflow_name: str | None = None
+    name: str,
+    code_only: bool,
+    limit: int,
+    created_at: DateRange | None = None,
+    workflow_name: str | None = None,
 ) -> None:
     load_dotenv()
 
     base = get_main_branch(name)
 
-    repository_histories = get_repository_history(name, base, limit)
+    repository_histories = get_repository_history(name, base, limit, created_at)
 
     # CircleCI
     deployment_histories = []

depwatch/date_utils.py

@@ -0,0 +1,47 @@
+from datetime import date
+
+
+class DateRange:
+    def __init__(
+        self,
+        start: date | None = None,
+        end: date | None = None,
+    ):
+        self.start = start
+        self.end = end
+
+    @staticmethod
+    def from_str(string: str):
+        try:
+            start_date_str, end_date_str = string.split("..")
+            start_date = (
+                date.fromisoformat(start_date_str) if start_date_str != "" else None
+            )
+            end_date = date.fromisoformat(end_date_str) if end_date_str != "" else None
+            return DateRange(start_date, end_date)
+        except ValueError:
+            raise ValueError(
+                "Invalid date range string format. Expected 'YYYY-MM-DD..YYYY-MM-DD'."
+            )
+
+    def __str__(self):
+        start_str = self.start if self.start is not None else ""
+        end_str = self.end if self.end is not None else ""
+        return f"{start_str}..{end_str}"
+
+    def __eq__(self, other: object) -> bool:
+        if isinstance(other, type(self)):
+            return (self.start, self.end) == (other.start, other.end)
+
+        return False
+
+
+def convert_date_range_to_str_for_search_query(date_range: DateRange) -> str | None:
+    if date_range.start is not None and date_range.end is not None:
+        return str(date_range)
+    elif date_range.start is not None and date_range.end is None:
+        return f">={str(date_range.start)}"
+    elif date_range.start is None and date_range.end is not None:
+        return f"<={str(date_range.end)}"
+    else:
+        return None

depwatch/main.py

@@ -2,6 +2,7 @@
 from rich import print
 
 from depwatch.command import generate_histories
+from depwatch.date_utils import DateRange
 
 
 app = typer.Typer()
@@ -14,12 +15,17 @@ def main(
         False, help="do not retrieve deployment histories from CI"
     ),
     limit: int = typer.Option(100, help="count limit for retrieving history"),
+    created_at: DateRange = typer.Option(
+        None,
+        parser=lambda str: DateRange.from_str(str),
+        help="The range of the period when the pull requests were created",
+    ),
     workflow_name: str = typer.Option(
         None,
         help="The workflow name of the CI to be obtained. This is useful when there are multiple workflows triggered by commits to the main branch.",
     ),
 ):
-    generate_histories(name, code_only, limit, workflow_name)
+    generate_histories(name, code_only, limit, created_at, workflow_name)
     print(":sparkles::sparkles: [green]Completed![/green] :sparkles::sparkles:")
 
 

depwatch/repository.py

@@ -1,6 +1,7 @@
 from github import Github
 import os
 from datetime import datetime, timezone
+from depwatch.date_utils import DateRange, convert_date_range_to_str_for_search_query
 from depwatch.exception import DepwatchException
 from depwatch.history import RepositoryHistory
 
@@ -19,12 +20,22 @@ def get_main_branch(name: str) -> str:
         raise DepwatchException("'main' or 'master' branch was not found")
 
 
-def get_repository_history(name: str, base: str, limit: int) -> list[RepositoryHistory]:
+def get_repository_history(
+    name: str, base: str, limit: int, created_at: DateRange | None = None
+) -> list[RepositoryHistory]:
     histories = []
 
     gh = Github(os.environ.get("GITHUB_ACCESS_TOKEN"), per_page=100)
     repo = gh.get_repo(name)
-    pulls = repo.get_pulls(state="closed", base=base)[:limit]
+    created_at_query = (
+        f"created:{convert_date_range_to_str_for_search_query(created_at)}"
+        if created_at is not None
+        else ""
+    )
+    issues = gh.search_issues(
+        f"repo:{name} type:pr is:merged {created_at_query}", "created", "desc"
+    )[:limit]
+    pulls = [i.as_pull_request() for i in issues]
 
     for p in pulls:
         if p.merged_at is None:

tests/test_command.py

@@ -2,6 +2,7 @@
 
 from depwatch import command
 from depwatch.command import generate_histories
+from depwatch.date_utils import DateRange
 
 
 class TestCommand:
@@ -20,14 +21,23 @@ def test_generate_histories(
         get_deployment_history_mock: Mock,
         write_histories_mock: Mock,
     ):
-        generate_histories("hamakou108/my_project", False, 100)
+        generate_histories(
+            "hamakou108/my_project",
+            False,
+            100,
+            DateRange.from_str("2023-01-01..2023-03-31"),
+            "workflow",
+        )
 
         get_main_branch_mock.assert_called_once_with("hamakou108/my_project")
         get_repository_history_mock.assert_called_once_with(
-            "hamakou108/my_project", "main", 100
+            "hamakou108/my_project",
+            "main",
+            100,
+            DateRange.from_str("2023-01-01..2023-03-31"),
         )
         convert_repository_history_to_workflow_ids_mock.assert_called_once_with(
-            [], None
+            [], "workflow"
         )
         get_deployment_history_mock.assert_called_once_with([])
         write_histories_mock.assert_called()
@@ -43,6 +53,6 @@ def test_generate_histories_with_code_only(
         get_deployment_history_mock: Mock,
         write_histories_mock: Mock,
     ):
-        generate_histories("hamakou108/my_project", True, 100, None)
+        generate_histories("hamakou108/my_project", True, 100, None, None)
 
         get_deployment_history_mock.assert_not_called()

tests/test_date_utils.py

@@ -0,0 +1,43 @@
+from datetime import date
+from depwatch.date_utils import DateRange, convert_date_range_to_str_for_search_query
+
+
+class TestDateUtils:
+    def test_date_range_from_str(self):
+        datasets = [
+            "2023-01-01..2023-03-31",
+            "2023-01-01..",
+            "..2023-03-31",
+        ]
+
+        for string in datasets:
+            date_range = DateRange.from_str(string)
+
+            assert str(date_range) == string
+
+    def test_date_range_from_str_when_invalid_string_is_provided(self):
+        try:
+            DateRange.from_str("2023-01-01")
+            assert False
+        except ValueError as e:
+            assert (
+                str(e)
+                == "Invalid date range string format. Expected 'YYYY-MM-DD..YYYY-MM-DD'."
+            )
+
+    def test_date_convert_date_range_to_str_for_search_query(self):
+        datasets = [
+            [
+                date.fromisoformat("2023-01-01"),
+                date.fromisoformat("2023-03-31"),
+                "2023-01-01..2023-03-31",
+            ],
+            [date.fromisoformat("2023-01-01"), None, ">=2023-01-01"],
+            [None, date.fromisoformat("2023-03-31"), "<=2023-03-31"],
+            [None, None, None],
+        ]
+
+        for start, end, string in datasets:
+            date_range = DateRange(start, end)
+
+            assert convert_date_range_to_str_for_search_query(date_range) == string

tests/test_main.py

@@ -1,4 +1,5 @@
 from unittest.mock import patch, Mock
+from depwatch.date_utils import DateRange
 from depwatch.main import app
 from typer.testing import CliRunner
 
@@ -14,7 +15,7 @@ def test_main(self, mock_generate_histories: Mock):
         assert "✨✨ Completed! ✨✨" in result.stdout
         assert result.exit_code == 0
         mock_generate_histories.assert_called_once_with(
-            "hamakou108/my_project", False, 100, None
+            "hamakou108/my_project", False, 100, None, None
         )
 
     @patch("depwatch.main.generate_histories")
@@ -26,12 +27,19 @@ def test_main_with_all_args(self, mock_generate_histories: Mock):
                 "--code-only",
                 "--limit",
                 "10",
+                "--created-at",
+                "2023-01-01..2023-03-31",
                 "--workflow-name",
                 "deploy",
             ],
         )
 
         assert result.exit_code == 0
+
         mock_generate_histories.assert_called_once_with(
-            "hamakou108/my_project", True, 10, "deploy"
+            "hamakou108/my_project",
+            True,
+            10,
+            DateRange.from_str("2023-01-01..2023-03-31"),
+            "deploy",
         )

tests/test_repository.py

@@ -2,6 +2,7 @@
 import secrets
 from unittest.mock import patch, Mock, MagicMock
 from depwatch import repository
+from depwatch.date_utils import DateRange
 from depwatch.exception import DepwatchException
 
 
@@ -36,16 +37,27 @@ def test_get_main_branch_with_other_branch(self, mock_Github: Mock):
 
     @patch("depwatch.repository.Github")
     def test_get_repository_history(self, mock_Github: Mock):
-        mock_repo = MagicMock()
-        mock_repo.get_pulls.return_value = [
-            self.create_mock_pull(),
-            self.create_mock_pull(),
-            self.create_mock_pull(),
+        mock_Github.return_value.search_issues.return_value = [
+            self.create_mock_issue(),
+            self.create_mock_issue(),
+            self.create_mock_issue(),
         ]
+        mock_repo = MagicMock()
         mock_repo.get_commit.return_value = self.create_mock_commit()
         mock_Github.return_value.get_repo.return_value = mock_repo
 
-        result = repository.get_repository_history("hamakou108/my_project", "main", 100)
+        result = repository.get_repository_history(
+            "hamakou108/my_project",
+            "main",
+            100,
+            DateRange.from_str("2023-01-01..2023-03-31"),
+        )
+
+        mock_Github.return_value.search_issues.assert_called_once_with(
+            "repo:hamakou108/my_project type:pr is:merged created:2023-01-01..2023-03-31",
+            "created",
+            "desc",
+        )
 
         assert len(result) == 3
         assert result[0].first_committed_at == datetime(2021, 1, 1, tzinfo=timezone.utc)
@@ -54,36 +66,58 @@ def test_get_repository_history(self, mock_Github: Mock):
         assert len(result[0].check_runs) == 1
 
     @patch("depwatch.repository.Github")
-    def test_get_repository_history_if_a_pull_request_with_no_merge_commit_is_included(
+    def test_get_repository_history_when_the_limit_is_specified(
         self, mock_Github: Mock
     ):
+        mock_Github.return_value.search_issues.return_value = [
+            self.create_mock_issue(),
+            self.create_mock_issue(),
+            self.create_mock_issue(),
+        ]
         mock_repo = MagicMock()
-        mock_repo.get_pulls.return_value = [
-            self.create_mock_pull(False),
+        mock_repo.get_commit.return_value = self.create_mock_commit(False)
+        mock_Github.return_value.get_repo.return_value = mock_repo
+
+        result = repository.get_repository_history("hamakou108/my_project", "main", 2)
+
+        assert len(result) == 2
+
+    @patch("depwatch.repository.Github")
+    def test_get_repository_history_when_the_created_at_is_not_specified(
+        self, mock_Github: Mock
+    ):
+        mock_Github.return_value.search_issues.return_value = [
+            self.create_mock_issue(),
+            self.create_mock_issue(),
+            self.create_mock_issue(),
         ]
+        mock_repo = MagicMock()
+        mock_repo.get_commit.return_value = self.create_mock_commit(False)
         mock_Github.return_value.get_repo.return_value = mock_repo
 
         result = repository.get_repository_history("hamakou108/my_project", "main", 100)
 
-        assert len(result) == 0
+        mock_Github.return_value.search_issues.assert_called_once_with(
+            "repo:hamakou108/my_project type:pr is:merged ", "created", "desc"
+        )
+
+        assert len(result) == 3
 
     @patch("depwatch.repository.Github")
     def test_get_repository_history_if_a_merge_commit_with_no_check_runs_is_included(
         self, mock_Github: Mock
     ):
-        mock_repo = MagicMock()
-        mock_repo.get_pulls.return_value = [
-            self.create_mock_pull(),
+        mock_Github.return_value.search_issues.return_value = [
+            self.create_mock_issue(),
         ]
+        mock_repo = MagicMock()
         mock_repo.get_commit.return_value = self.create_mock_commit(False)
         mock_Github.return_value.get_repo.return_value = mock_repo
 
         result = repository.get_repository_history("hamakou108/my_project", "main", 100)
 
         assert len(result) == 1
         assert len(result[0].check_runs) == 0
-        # assert result[0].check_run_app_slug == None
-        # assert result[0].check_run_external_id == None
 
     def create_mock_repo_with_branches(self, branches: list[str]):
         mock_repo = MagicMock()
@@ -116,7 +150,7 @@ def create_mock_commit(self, has_check_runs: bool = True):
 
         return mock_commit
 
-    def create_mock_pull(self, is_merged: bool = True):
+    def create_mock_pull(self):
         mock_pull = MagicMock(spec=["get_commits", "merged_at", "merge_commit_sha"])
         mock_commits = []
         for i in range(3):
@@ -126,9 +160,13 @@ def create_mock_pull(self, is_merged: bool = True):
             )
             mock_commits.append(mock_commit)
         mock_pull.get_commits.return_value = mock_commits
-        mock_pull.merged_at = (
-            datetime(2021, 1, 3, tzinfo=timezone.utc) if is_merged else None
-        )
-        mock_pull.merge_commit_sha = secrets.token_hex(20) if is_merged else None
+        mock_pull.merged_at = datetime(2021, 1, 3, tzinfo=timezone.utc)
+        mock_pull.merge_commit_sha = secrets.token_hex(20)
 
         return mock_pull
+
+    def create_mock_issue(self):
+        mock_issue = MagicMock()
+        mock_issue.as_pull_request.return_value = self.create_mock_pull()
+
+        return mock_issue