Infers and autocompletes the --from repo when possible (#59)

Most folks will likely either be working outside of a git repo, or
within a git repo with an origin at Github.  In these cases, we can just
assume/default to the fact that they likely want _this_ repo to be their
code storage.  Here we add autocompletion and defaulting to guess the
repo when it's not provided.

src/prefect_cloud/cli/completions.py

@@ -3,6 +3,7 @@
 from pathlib import Path
 
 from prefect_cloud.auth import get_cloud_urls_without_login, sync_cloud_client
+from prefect_cloud.github import get_local_repo_urls
 
 COMPLETION_CACHE = Path.home() / ".prefect" / "prefect-cloud-completions.json"
 CACHE_TTL = 86400
@@ -52,3 +53,7 @@ def complete_deployment(incomplete: str) -> list[str]:
             json.dump({"deployment_names": deployment_names}, f)
 
     return [name for name in deployment_names if name.startswith(incomplete)]
+
+
+def complete_repo(incomplete: str) -> list[str]:
+    return [url for url in get_local_repo_urls() if url.startswith(incomplete)]

src/prefect_cloud/cli/root.py

@@ -14,10 +14,7 @@
     process_key_value_pairs,
 )
 from prefect_cloud.dependencies import get_dependencies
-from prefect_cloud.github import (
-    FileNotFound,
-    GitHubRepo,
-)
+from prefect_cloud.github import FileNotFound, GitHubRepo, infer_repo_url
 from prefect_cloud.schemas.objects import (
     CronSchedule,
     DeploymentSchedule,
@@ -45,11 +42,14 @@ async def deploy(
         ...,
         "--from",
         "-f",
+        default_factory=infer_repo_url,
+        autocompletion=completions.complete_repo,
         help=(
             "GitHub repository URL. e.g.\n\n"
             "• Repo: github.com/owner/repo\n\n"
             "• Specific branch: github.com/owner/repo/tree/<branch>\n\n"
             "• Specific commit: github.com/owner/repo/tree/<commit-sha>\n\n"
+            "If not provided, the repository of the current directory will be used."
         ),
         rich_help_panel="Source",
         show_default=False,

src/prefect_cloud/github.py

@@ -1,9 +1,12 @@
+import subprocess
 from dataclasses import dataclass
 from typing import Any
 from urllib.parse import urlparse
 
 from httpx import AsyncClient
 
+from prefect_cloud.cli.utilities import exit_with_error
+
 
 class FileNotFound(Exception):
     pass
@@ -111,3 +114,72 @@ def to_pull_step(self, credentials_block: str | None = None) -> dict[str, Any]:
             )
 
         return {"prefect.deployments.steps.git_clone": pull_step_kwargs}
+
+
+def translate_to_http(url: str) -> str:
+    """
+    Translate a git URL to an HTTP URL.
+    """
+    url = url.strip()
+
+    if url.startswith("[email protected]:"):
+        url = "https://github.com/" + url.split("[email protected]:")[1]
+
+    if url.endswith(".git"):
+        url = url.removesuffix(".git")
+
+    return url
+
+
+def infer_repo_url() -> str:
+    """
+    Infer the repository URL from the current directory.
+    """
+    try:
+        result = subprocess.run(
+            ["git", "remote", "get-url", "origin"],
+            capture_output=True,
+            text=True,
+            check=True,
+        )
+        url = result.stdout.strip()
+
+        url = translate_to_http(url)
+
+        if not url.startswith("https://github.com"):
+            raise ValueError("Repository URL must be from github.com")
+
+        return url
+
+    except (subprocess.CalledProcessError, ValueError):
+        exit_with_error(
+            "No repository specified, and this directory doesn't appear to be a "
+            "GitHub repository.  Specify --from to indicate where Prefect Cloud will "
+            "download your code from."
+        )
+
+
+def get_local_repo_urls() -> list[str]:
+    """
+    Get all local repository URLs from the current directory.
+    """
+    try:
+        remotes = subprocess.run(
+            ["git", "remote", "show"],
+            capture_output=True,
+            text=True,
+            check=True,
+        )
+        all_urls: list[str] = []
+        for remote in remotes.stdout.splitlines():
+            result = subprocess.run(
+                ["git", "remote", "get-url", remote],
+                capture_output=True,
+                text=True,
+                check=True,
+            )
+            urls = [translate_to_http(url) for url in result.stdout.splitlines()]
+            all_urls += [url for url in urls if url.startswith("https://github.com")]
+        return all_urls
+    except (subprocess.CalledProcessError, ValueError):
+        return []

tests/test_github.py

@@ -1,7 +1,18 @@
+import os
+import subprocess
+import tempfile
+from pathlib import Path
+
+import click.exceptions
 import pytest
 from httpx import Response
 
-from prefect_cloud.github import FileNotFound, GitHubRepo
+from prefect_cloud.github import (
+    FileNotFound,
+    GitHubRepo,
+    get_local_repo_urls,
+    infer_repo_url,
+)
 
 
 class TestGitHubRepo:
@@ -198,3 +209,132 @@ def test_to_pull_step_with_credentials(self):
                 "access_token": "{{ prefect.blocks.secret.test-creds }}",
             }
         }
+
+
+@pytest.fixture
+def git_repo(tmp_path: Path) -> Path:
+    """Create a temporary git repository."""
+    os.chdir(tmp_path)
+    subprocess.run(["git", "init"], check=True, capture_output=True)
+    return tmp_path
+
+
+class TestInferRepoUrl:
+    def test_infers_https_url(self, git_repo: Path):
+        subprocess.run(
+            [
+                "git",
+                "remote",
+                "add",
+                "origin",
+                "https://github.com/ExampleOwner/example-repo",
+            ],
+            check=True,
+            capture_output=True,
+        )
+
+        assert infer_repo_url() == "https://github.com/ExampleOwner/example-repo"
+
+    def test_infers_ssh_url(self, git_repo: Path):
+        subprocess.run(
+            [
+                "git",
+                "remote",
+                "add",
+                "origin",
+                "[email protected]:ExampleOwner/example-repo.git",
+            ],
+            check=True,
+            capture_output=True,
+        )
+
+        assert infer_repo_url() == "https://github.com/ExampleOwner/example-repo"
+
+    def test_exits_when_not_git_repo(self):
+        with tempfile.TemporaryDirectory() as temp_dir:
+            os.chdir(temp_dir)
+
+            with pytest.raises(click.exceptions.Exit):
+                infer_repo_url()
+
+    def test_exits_when_not_github_url(self, git_repo: Path):
+        subprocess.run(
+            [
+                "git",
+                "remote",
+                "add",
+                "origin",
+                "https://gitlab.com/ExampleOwner/example-repo",
+            ],
+            check=True,
+            capture_output=True,
+        )
+
+        with pytest.raises(click.exceptions.Exit):
+            infer_repo_url()
+
+
+class TestGetLocalRepoUrls:
+    def test_returns_empty_list_when_not_git_repo(self):
+        with tempfile.TemporaryDirectory() as temp_dir:
+            os.chdir(temp_dir)
+
+            assert get_local_repo_urls() == []
+
+    def test_returns_github_urls(self, git_repo: Path):
+        remotes = [
+            ("origin", "https://github.com/ExampleOwner/example-repo"),
+            ("upstream", "https://github.com/UpstreamOwner/example-repo"),
+        ]
+
+        for name, url in remotes:
+            subprocess.run(
+                ["git", "remote", "add", name, url],
+                check=True,
+                capture_output=True,
+            )
+
+        urls = get_local_repo_urls()
+        assert len(urls) == 2
+        assert set(urls) == {remote[1] for remote in remotes}
+
+    def test_filters_non_github_urls(self, git_repo: Path):
+        remotes = [
+            ("origin", "https://github.com/ExampleOwner/example-repo"),
+            ("gitlab", "https://gitlab.com/ExampleOwner/example-repo"),
+            ("upstream", "https://github.com/UpstreamOwner/example-repo"),
+        ]
+
+        for name, url in remotes:
+            subprocess.run(
+                ["git", "remote", "add", name, url],
+                check=True,
+                capture_output=True,
+            )
+
+        urls = get_local_repo_urls()
+        assert len(urls) == 2
+        assert set(urls) == {
+            "https://github.com/ExampleOwner/example-repo",
+            "https://github.com/UpstreamOwner/example-repo",
+        }
+
+    def test_translates_ssh_urls(self, git_repo: Path):
+        remotes = [
+            ("origin", "[email protected]:ExampleOwner/example-repo.git"),
+            ("upstream", "https://github.com/UpstreamOwner/example-repo"),
+        ]
+
+        for name, url in remotes:
+            subprocess.run(
+                ["git", "remote", "add", name, url],
+                check=True,
+                capture_output=True,
+            )
+
+        urls = get_local_repo_urls()
+        assert len(urls) == 2
+        assert set(urls) == {
+            "https://github.com/ExampleOwner/example-repo",
+            "https://github.com/UpstreamOwner/example-repo",
+        }