wehub-resource-sync
98 lines
3.3 KiB
Python
98 lines
3.3 KiB
Python
import logging
|
|
import os
|
|
from urllib.parse import urlsplit, urlunsplit
|
|
|
|
_logger = logging.getLogger(__name__)
|
|
|
|
|
|
def _strip_credentials_from_url(url: str) -> str:
|
|
"""
|
|
Strip any embedded userinfo (username/password) from a URL so it's safe to record as a tag.
|
|
|
|
HTTP(S) and similar remotes can carry credentials in the form
|
|
``https://user:token@host/path``; MLflow records repo URLs as run tags that are stored and
|
|
displayed to users, so we drop the userinfo before exposing them. URLs without a scheme
|
|
(e.g. SSH-style ``git@github.com:org/repo.git``) or without userinfo are returned unchanged.
|
|
"""
|
|
parsed = urlsplit(url)
|
|
if not parsed.scheme or "@" not in parsed.netloc:
|
|
return url
|
|
_, _, host_port = parsed.netloc.rpartition("@")
|
|
return urlunsplit((parsed.scheme, host_port, parsed.path, parsed.query, parsed.fragment))
|
|
|
|
|
|
def get_git_repo_url(path: str) -> str | None:
|
|
"""
|
|
Obtains the url of the git repository associated with the specified path,
|
|
returning ``None`` if the path does not correspond to a git repository. Any embedded
|
|
credentials (e.g. tokens in HTTPS remotes) are stripped before the URL is returned.
|
|
"""
|
|
try:
|
|
from git import Repo
|
|
except ImportError as e:
|
|
_logger.warning(
|
|
"Failed to import Git (the Git executable is probably not on your PATH),"
|
|
" so Git repository URL is not available. Error: %s",
|
|
e,
|
|
)
|
|
return None
|
|
|
|
try:
|
|
if os.path.isfile(path):
|
|
path = os.path.dirname(os.path.abspath(path))
|
|
repo = Repo(path, search_parent_directories=True)
|
|
url = next((remote.url for remote in repo.remotes), None)
|
|
return _strip_credentials_from_url(url) if url is not None else None
|
|
except Exception:
|
|
return None
|
|
|
|
|
|
def get_git_commit(path: str) -> str | None:
|
|
"""
|
|
Obtains the hash of the latest commit on the current branch of the git repository associated
|
|
with the specified path, returning ``None`` if the path does not correspond to a git
|
|
repository.
|
|
"""
|
|
try:
|
|
from git import Repo
|
|
except ImportError as e:
|
|
_logger.warning(
|
|
"Failed to import Git (the Git executable is probably not on your PATH),"
|
|
" so Git SHA is not available. Error: %s",
|
|
e,
|
|
)
|
|
return None
|
|
try:
|
|
if os.path.isfile(path):
|
|
path = os.path.dirname(os.path.abspath(path))
|
|
repo = Repo(path, search_parent_directories=True)
|
|
if path in repo.ignored(path):
|
|
return None
|
|
return repo.head.commit.hexsha
|
|
except Exception:
|
|
return None
|
|
|
|
|
|
def get_git_branch(path: str) -> str | None:
|
|
"""
|
|
Obtains the name of the current branch of the git repository associated with the specified
|
|
path, returning ``None`` if the path does not correspond to a git repository.
|
|
"""
|
|
try:
|
|
from git import Repo
|
|
except ImportError as e:
|
|
_logger.warning(
|
|
"Failed to import Git (the Git executable is probably not on your PATH),"
|
|
" so Git branch is not available. Error: %s",
|
|
e,
|
|
)
|
|
return None
|
|
|
|
try:
|
|
if os.path.isfile(path):
|
|
path = os.path.dirname(os.path.abspath(path))
|
|
repo = Repo(path, search_parent_directories=True)
|
|
return repo.active_branch.name
|
|
except Exception:
|
|
return None
|