Pour tout problème contactez-nous par mail : support@froggit.fr | La FAQ :grey_question: | Rejoignez-nous sur le Chat :speech_balloon:

Skip to content
Snippets Groups Projects
Commit fc040bf7 authored by Dorian Turba's avatar Dorian Turba
Browse files

move everything in release service

parent 32a04e64
No related branches found
No related tags found
No related merge requests found
Hide whitespace changes
Inline Side-by-side
src/release_by_changelog/cmds/release.py
+ 9
− 52
View file @ fc040bf7
from __future__ import annotations
import pathlib
import typing
import gitlab.v4.objects
import typer
import release_by_changelog.logging as logging
from release_by_changelog.app import app
from release_by_changelog.services.changelog import find_changelog_file, last_changelog
from release_by_changelog.services.gitlab import get_project
from release_by_changelog.services.write import post
from release_by_changelog.typings_ import ChangelogEntry
from release_by_changelog.services.release import (
find_changelog_file,
get_project,
get_token,
last_changelog,
publish,
)
@app.command()
......@@ -62,52 +65,6 @@ def release(
publish(changelog_entry, project_, project_name, ref, tag_only)
def get_token(ci_job_token: str | None, token: str | None) -> str:
"""
Get the token from the environment variables or the CLI.
:raises typer.Exit: If no token is provided.
"""
if token is not None:
return token
if ci_job_token is not None:
return ci_job_token
logging.error(
logging.err_panel(
"You need to provide a PRIVATE_TOKEN or a CI_JOB_TOKEN",
title="Error: Missing token",
)
)
raise typer.Exit(code=1)
def publish(
changelog_entry: ChangelogEntry,
project_: gitlab.v4.objects.Project,
project_name: str,
ref: str,
tag_only: bool,
) -> None:
logging.info(
f"Creating release [bold cyan]{changelog_entry.version}[/bold cyan] for "
f"project [bold cyan]{project_name}[/bold cyan]"
)
target: dict[bool, typing.Literal["tags", "releases"]] = {
True: "tags",
False: "releases",
}
post(
project_,
changelog_entry,
ref,
target[tag_only],
)
if __name__ == "__main__":
release(
project="1484",
......
src/release_by_changelog/services/gitlab.py deleted 100644 → 0
+ 0
− 61
View file @ 32a04e64
from __future__ import annotations
import functools
import http
import gitlab.v4.objects
import typer
from release_by_changelog import logging
@functools.cache
def get_project(
token: str,
host: str,
project: str,
) -> tuple[gitlab.v4.objects.Project, str]:
logging.info(
f"Retrieving project [bold cyan]{project}[/bold cyan] from "
f"[bold cyan]{host}[/bold cyan]"
)
url = f"https://{host}"
gl = gitlab.Gitlab(url=url, oauth_token=token)
try:
project_ = gl.projects.get(project)
except gitlab.GitlabAuthenticationError as e:
if e.response_code == http.HTTPStatus.UNAUTHORIZED:
logging.error(
logging.err_panel(
"Possible remediation:\n"
"\t- Check if the provided token is correct.\n"
"\t- Check if the provided token has the correct permissions and "
"scope, is not expired or revoked.",
title="Error: Unauthorized",
)
)
raise typer.Exit(code=1)
raise e
except gitlab.GitlabGetError as e:
if e.response_code == http.HTTPStatus.NOT_FOUND:
logging.error(
logging.err_panel(
"Possible remediation:\n"
"\t- Provide the project ID. To find the project ID, go to the "
"project page, click on the three vertical dot button on the top "
"right corner and press the 'Copy project ID: XXXX' button.\n"
"\t- Provide the full path to the project. To find it, go to the "
"project page, check at the url and copy the path after the host. "
"If the following link doesn't point to your project, you could "
f"have the namespace wrong: 'https://{host}/{project}'\n"
"\t- Check if the host is correct. If you are using a self-hosted "
"GitLab, you need to provide the correct host. Current host: "
f"'https://{host}', is that where your project is hosted?\n",
title="Error: Project not found.",
)
)
raise typer.Exit(code=1)
raise e
project_name = f"{project_.namespace.get('full_path')}/{project_.name}"
logging.success(f"Project found: [bold cyan]{project_name}[bold cyan]")
return project_, project_name
src/release_by_changelog/services/read.py deleted 100644 → 0
+ 0
− 0
View file @ 32a04e64
src/release_by_changelog/services/changelog.py → src/release_by_changelog/services/release.py
+ 260
− 0
View file @ fc040bf7
from __future__ import annotations
import contextlib
import functools
import http
import os
import pathlib
import re
import typing
import gitlab.v4
import gitlab.v4.objects
import typer
from release_by_changelog import logging
from release_by_changelog.services.gitlab import get_project
from release_by_changelog.typings_ import ChangelogEntry
regex: typing.Final = re.compile(
r"^## \[(?P<version>(?P<major>0|[1-9]\d*)\.(?P<minor>0|[1-9]\d*)\.(?P<patch>0|[1-9]"
r"\d*)(?:-(?P<prerelease>(?:0|[1-9]\d*|\d*[a-zA-Z-][0-9a-zA-Z-]*)(?:\.(?:0|[1-9]\d*"
r"|\d*[a-zA-Z-][0-9a-zA-Z-]*))*))?(?:\+(?P<buildmetadata>[0-9a-zA-Z-]+(?:\.[0-9a-zA"
r"-Z-]+)*))?)\]"
)
def publish(
changelog_entry: ChangelogEntry,
project: gitlab.v4.objects.Project,
project_name: str,
ref: str,
tag_only: bool,
) -> None:
logging.info(
f"Creating release [bold cyan]{changelog_entry.version}[/bold cyan] for "
f"project [bold cyan]{project_name}[/bold cyan]"
)
def _extract_last_version(f: typing.TextIO) -> str:
data = {
"tag_name": changelog_entry.version,
"ref": ref,
}
try:
getattr(project, TARGET_RELEASE[tag_only]).create(data)
except gitlab.GitlabAuthenticationError as e:
if e.response_code == http.HTTPStatus.UNAUTHORIZED:
logging.error(
logging.err_panel(
"Possible remediation:\n"
"\t- Check if the provided token is correct.\n"
"\t- Check if the provided token has the correct permissions and "
"scope, is not expired or revoked.",
title="Error: Unauthorized",
)
)
raise typer.Exit(code=1)
raise e
except gitlab.GitlabCreateError as e:
if e.response_code == http.HTTPStatus.CONFLICT:
logging.error(
logging.err_panel(
f"It looks like the {TARGET_RELEASE[tag_only]} [bold cyan]"
f"{changelog_entry.version}[/bold cyan] already exists.\n"
"Possible remediation: Bump the version in the changelog file.",
title=f"Error: {TARGET_RELEASE[tag_only].capitalize()} already "
f"exists",
)
)
raise typer.Exit(code=1)
raise e
logging.success(f"Release created: {changelog_entry.version}")
TARGET_RELEASE: typing.Final[dict[bool, typing.Literal["tags", "releases"]]] = {
True: "tags",
False: "releases",
}
def get_token(ci_job_token: str | None, token: str | None) -> str:
"""
Get the token from the environment variables or the CLI.
:raises typer.Exit: If no token is provided.
"""
if token is not None:
return token
if ci_job_token is not None:
return ci_job_token
logging.error(
logging.err_panel(
"You need to provide a PRIVATE_TOKEN or a CI_JOB_TOKEN",
title="Error: Missing token",
)
)
raise typer.Exit(code=1)
def _extract_body(f: typing.TextIO) -> str:
body = []
for lines in f:
matches = regex.finditer(lines)
try:
match = next(matches)
except StopIteration:
continue
return match.group("version")
raise ValueError("No changelog entry found")
with contextlib.suppress(StopIteration):
next(matches)
break
body.append(lines)
return "".join(body)
def last_changelog(changelog_path: pathlib.Path) -> ChangelogEntry:
"""Extract the last changelog entry from the changelog file."""
logging.info(f"Processing [bold cyan]{changelog_path}[/bold cyan]")
with changelog_path.open() as f:
version = _extract_last_version(f)
body = _extract_body(f)
logging.success(f"Found changelog entry: [bold cyan]{version}[/bold cyan]")
return ChangelogEntry(version=version, body=body)
def _get_remote_changelog_file(
changelog_path: pathlib.Path,
project_: gitlab.v4.objects.Project,
ref: str,
) -> gitlab.v4.objects.ProjectFile:
try:
changelog_file: gitlab.v4.objects.ProjectFile = project_.files.get(
file_path=str(changelog_path), ref=ref
)
except gitlab.GitlabGetError as e:
if e.response_code == http.HTTPStatus.NOT_FOUND:
logging.error(
f"[bold red]Changelog file {changelog_path} not found in the remote "
f"project files[/bold red]"
)
raise typer.Exit(code=1)
raise e
logging.success(
f"[bold cyan]{changelog_path}[/bold cyan] found in the remote project files"
)
return changelog_file
def _save_remote_changelog_file(
changelog_file: gitlab.v4.objects.ProjectFile, changelog_path: pathlib.Path
) -> pathlib.Path:
"""
Save the remote changelog file to a temporary file.
:raises NotImplementedError: If the OS is not supported.
"""
if os.name == "posix":
tmp_file = pathlib.Path(f"/tmp/{changelog_path.name}")
elif os.name == "nt":
temp_path = pathlib.Path(os.environ["Temp"])
tmp_file = temp_path / changelog_path.name
else:
raise NotImplementedError(f"OS {os.name} not supported")
tmp_file.write_bytes(changelog_file.decode())
return tmp_file
def find_changelog_file(
......@@ -78,55 +179,82 @@ def find_changelog_file(
return tmp_file
def _save_remote_changelog_file(
changelog_file: gitlab.v4.objects.ProjectFile, changelog_path: pathlib.Path
) -> pathlib.Path:
"""
Save the remote changelog file to a temporary file.
def last_changelog(changelog_path: pathlib.Path) -> ChangelogEntry:
"""Extract the last changelog entry from the changelog file."""
logging.info(f"Processing [bold cyan]{changelog_path}[/bold cyan]")
with changelog_path.open() as f:
version = _extract_last_version(f)
body = _extract_body(f)
logging.success(f"Found changelog entry: [bold cyan]{version}[/bold cyan]")
return ChangelogEntry(version=version, body=body)
:raises NotImplementedError: If the OS is not supported.
"""
if os.name == "posix":
tmp_file = pathlib.Path(f"/tmp/{changelog_path.name}")
elif os.name == "nt":
temp_path = pathlib.Path(os.environ["Temp"])
tmp_file = temp_path / changelog_path.name
else:
raise NotImplementedError(f"OS {os.name} not supported")
tmp_file.write_bytes(changelog_file.decode())
return tmp_file
def _extract_last_version(f: typing.TextIO) -> str:
for lines in f:
matches = regex.finditer(lines)
try:
match = next(matches)
except StopIteration:
continue
return match.group("version")
raise ValueError("No changelog entry found")
def _get_remote_changelog_file(
changelog_path: pathlib.Path,
project_: gitlab.v4.objects.Project,
ref: str,
) -> gitlab.v4.objects.ProjectFile:
regex: typing.Final = re.compile(
r"^## \[(?P<version>(?P<major>0|[1-9]\d*)\.(?P<minor>0|[1-9]\d*)\.(?P<patch>0|[1-9]"
r"\d*)(?:-(?P<prerelease>(?:0|[1-9]\d*|\d*[a-zA-Z-][0-9a-zA-Z-]*)(?:\.(?:0|[1-9]\d*"
r"|\d*[a-zA-Z-][0-9a-zA-Z-]*))*))?(?:\+(?P<buildmetadata>[0-9a-zA-Z-]+(?:\.[0-9a-zA"
r"-Z-]+)*))?)\]"
)
@functools.cache
def get_project(
token: str,
host: str,
project: str,
) -> tuple[gitlab.v4.objects.Project, str]:
logging.info(
f"Retrieving project [bold cyan]{project}[/bold cyan] from "
f"[bold cyan]{host}[/bold cyan]"
)
url = f"https://{host}"
gl = gitlab.Gitlab(url=url, oauth_token=token)
try:
changelog_file: gitlab.v4.objects.ProjectFile = project_.files.get(
file_path=str(changelog_path), ref=ref
)
project_ = gl.projects.get(project)
except gitlab.GitlabAuthenticationError as e:
if e.response_code == http.HTTPStatus.UNAUTHORIZED:
logging.error(
logging.err_panel(
"Possible remediation:\n"
"\t- Check if the provided token is correct.\n"
"\t- Check if the provided token has the correct permissions and "
"scope, is not expired or revoked.",
title="Error: Unauthorized",
)
)
raise typer.Exit(code=1)
raise e
except gitlab.GitlabGetError as e:
if e.response_code == http.HTTPStatus.NOT_FOUND:
logging.error(
f"[bold red]Changelog file {changelog_path} not found in the remote "
f"project files[/bold red]"
logging.err_panel(
"Possible remediation:\n"
"\t- Provide the project ID. To find the project ID, go to the "
"project page, click on the three vertical dot button on the top "
"right corner and press the 'Copy project ID: XXXX' button.\n"
"\t- Provide the full path to the project. To find it, go to the "
"project page, check at the url and copy the path after the host. "
"If the following link doesn't point to your project, you could "
f"have the namespace wrong: 'https://{host}/{project}'\n"
"\t- Check if the host is correct. If you are using a self-hosted "
"GitLab, you need to provide the correct host. Current host: "
f"'https://{host}', is that where your project is hosted?\n",
title="Error: Project not found.",
)
)
raise typer.Exit(code=1)
raise e
logging.success(
f"[bold cyan]{changelog_path}[/bold cyan] found in the remote project files"
)
return changelog_file
def _extract_body(f: typing.TextIO) -> str:
body = []
for lines in f:
matches = regex.finditer(lines)
with contextlib.suppress(StopIteration):
next(matches)
break
body.append(lines)
return "".join(body)
project_name = f"{project_.namespace.get('full_path')}/{project_.name}"
logging.success(f"Project found: [bold cyan]{project_name}[bold cyan]")
return project_, project_name
src/release_by_changelog/services/write.py deleted 100644 → 0
+ 0
− 50
View file @ 32a04e64
from __future__ import annotations
import http
import typing
import gitlab.v4.objects
import typer
from release_by_changelog import logging
from release_by_changelog.typings_ import ChangelogEntry
def post(
project: gitlab.v4.objects.Project,
changelog_entry: ChangelogEntry,
ref: str,
target: typing.Literal["tags", "releases"],
) -> None:
data = {
"tag_name": changelog_entry.version,
"ref": ref,
}
try:
getattr(project, target).create(data)
except gitlab.GitlabAuthenticationError as e:
if e.response_code == http.HTTPStatus.UNAUTHORIZED:
logging.error(
logging.err_panel(
"Possible remediation:\n"
"\t- Check if the provided token is correct.\n"
"\t- Check if the provided token has the correct permissions and "
"scope, is not expired or revoked.",
title="Error: Unauthorized",
)
)
raise typer.Exit(code=1)
raise e
except gitlab.GitlabCreateError as e:
if e.response_code == http.HTTPStatus.CONFLICT:
logging.error(
logging.err_panel(
f"It looks like the {target} [bold cyan]{changelog_entry.version}"
f"[/bold cyan] already exists.\n"
"Possible remediation: Bump the version in the changelog file.",
title=f"Error: {target.capitalize()} already exists",
)
)
raise typer.Exit(code=1)
raise e
logging.success(f"Release created: {changelog_entry.version}")
tests/conftest.py
+ 2
− 2
View file @ fc040bf7
......@@ -7,7 +7,7 @@ import pathlib
import click.testing
import pytest
import pytest_mock
import release_by_changelog.services.gitlab
import release_by_changelog.services.release
import typer.testing
from release_by_changelog.app import app as rbc_app
......@@ -330,4 +330,4 @@ def successful_gitlab_interaction(mocker: pytest_mock.MockFixture) -> None:
@pytest.fixture(autouse=True)
def clear_cached_functions() -> None:
release_by_changelog.services.gitlab.get_project.cache_clear()
release_by_changelog.services.release.get_project.cache_clear()
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment

Froggit est fièrement fabriqué avec amour par Lydra avec des Logiciels Libres et hébergé en France chez Scaleway | Status du service