in lib/metric-config-parser/metric_config_parser/config.py [0:0]
def as_of(self, timestamp: datetime) -> "ConfigCollection":
"""Get configs as they were at the provided timestamp."""
if timestamp is None:
return self
config_collection = None
# configs can be loaded from multiple different repos
for repo in self.repos:
# copy the original repo to a temporary path were we can go back in history
with TemporaryDirectory() as tmp_dir:
git_dir = Path(repo.repo.git_dir)
# copy over repo content, except for lib/*, which doesn't contain metric information
shutil.copytree(
git_dir.parent,
tmp_dir,
dirs_exist_ok=True,
ignore=shutil.ignore_patterns("lib*"),
)
tmp_repo = Repo(tmp_dir)
# find the commit that got added just before the `timestamp`
rev = "HEAD" # use the HEAD commit by default if no other commits exist
# make sure the most recent commit is checked out by checking out the main branch
tmp_repo.git.checkout(repo.main_branch)
# keep track of more recent commits to go back to in case invalid configs
# got checked into main that cannot be parsed
newer_commits: List[Commit] = []
# start iterating through all commits starting at HEAD
for commit in tmp_repo.iter_commits("HEAD"):
# check if the commit timestamp is older than the reference timestamp
if commit.committed_datetime <= timestamp:
rev = commit.hexsha
break
newer_commits.insert(0, commit)
if commit:
# if there is no commit that is older than the reference timestamp,
# use the oldest commit that we could find (= last commit)
rev = commit.hexsha
# checkout that commit
tmp_repo.git.checkout(rev)
# load configs as they were at the time of the commit
try:
configs = self.from_local_repo(
tmp_repo,
repo.path,
self.is_private,
repo.main_branch,
is_tmp_repo=True,
)
except Exception as e:
could_load_configs = False
# iterate through newer commits to find one that can be parsed
for newer_commit in newer_commits:
tmp_repo.git.checkout(newer_commit.hexsha)
try:
configs = self.from_local_repo(
tmp_repo,
repo.path,
self.is_private,
repo.main_branch,
is_tmp_repo=True,
)
could_load_configs = True
rev = newer_commit.hexsha
break
except Exception:
# continue searching
pass
if not could_load_configs:
# there is no newer commit, current state is broken
raise e
repo.commit_hash = rev
configs.repos = [repo] # point to the original repo, instead of the temporary one
if config_collection is None:
config_collection = configs
else:
config_collection.merge(configs)
if config_collection is None:
return self
return config_collection