prefect_bitbucket

Contains the credentials and repository submodules.

BitBucketCredentials

Bases: CredentialsBlock

Store BitBucket credentials to interact with private BitBucket repositories.

Attributes:

Name	Type	Description
token	Optional[SecretStr]	An access token to authenticate with BitBucket. This is required for accessing private repositories.
username	Optional[str]	Identification name unique across entire BitBucket site.
password	Optional[SecretStr]	The password to authenticate to BitBucket.
url	Optional[str]	The base URL of your BitBucket instance.

Examples:

Load stored BitBucket credentials:

from prefect_bitbucket import BitBucketCredentials
bitbucket_credentials_block = BitBucketCredentials.load("BLOCK_NAME")

Source code in prefect_bitbucket/credentials.py

class BitBucketCredentials(CredentialsBlock):
    """Store BitBucket credentials to interact with private BitBucket repositories.

    Attributes:
        token: An access token to authenticate with BitBucket. This is required
            for accessing private repositories.
        username: Identification name unique across entire BitBucket site.
        password: The password to authenticate to BitBucket.
        url: The base URL of your BitBucket instance.


    Examples:
        Load stored BitBucket credentials:
        ```python
        from prefect_bitbucket import BitBucketCredentials
        bitbucket_credentials_block = BitBucketCredentials.load("BLOCK_NAME")
        ```


    """

    _block_type_name = "BitBucket Credentials"
    _logo_url = "https://cdn.sanity.io/images/3ugk85nk/production/5d729f7355fb6828c4b605268ded9cfafab3ae4f-250x250.png"  # noqa

    token: Optional[SecretStr] = Field(
        title="Personal Access Token",
        default=None,
        description=(
            "A BitBucket Personal Access Token - required for private repositories."
        ),
        examples=["x-token-auth:my-token"],
    )
    username: Optional[str] = Field(
        default=None,
        description="Identification name unique across entire BitBucket site.",
    )
    password: Optional[SecretStr] = Field(
        default=None, description="The password to authenticate to BitBucket."
    )
    url: Optional[str] = Field(
        default=None,
        description="The base URL of a BitBucket instance. Leave blank for BitBucket Cloud.",
        examples=["https://api.bitbucket.org/"],
        title="URL",
    )

    @field_validator("username")
    def _validate_username(cls, value: str) -> str:
        """When username provided, will validate it."""
        pattern = "^[A-Za-z0-9_-]*$"

        if not re.match(pattern, value):
            raise ValueError(
                "Username must be alpha, num, dash and/or underscore only."
            )
        if not len(value) <= 30:
            raise ValueError("Username cannot be longer than 30 chars.")
        return value

    def get_client(
        self, client_type: Union[str, ClientType], **client_kwargs
    ) -> Union[Cloud, Bitbucket]:
        """Get an authenticated local or cloud Bitbucket client.

        Args:
            client_type: Whether to use a local or cloud client.

        Returns:
            An authenticated Bitbucket client.

        """
        # ref: https://atlassian-python-api.readthedocs.io/
        if isinstance(client_type, str):
            client_type = ClientType(client_type.lower())

        password = self.password.get_secret_value()
        input_client_kwargs = dict(
            url=self.url, username=self.username, password=password
        )
        input_client_kwargs.update(**client_kwargs)

        if client_type == ClientType.CLOUD:
            client = Cloud(**input_client_kwargs)
        else:
            client = Bitbucket(**input_client_kwargs)
        return client

get_client(client_type, **client_kwargs)

Get an authenticated local or cloud Bitbucket client.

Parameters:

Name	Type	Description	Default
client_type	Union[str, ClientType]	Whether to use a local or cloud client.	required

Returns:

Type	Description
Union[Cloud, Bitbucket]	An authenticated Bitbucket client.

Source code in prefect_bitbucket/credentials.py

def get_client(
    self, client_type: Union[str, ClientType], **client_kwargs
) -> Union[Cloud, Bitbucket]:
    """Get an authenticated local or cloud Bitbucket client.

    Args:
        client_type: Whether to use a local or cloud client.

    Returns:
        An authenticated Bitbucket client.

    """
    # ref: https://atlassian-python-api.readthedocs.io/
    if isinstance(client_type, str):
        client_type = ClientType(client_type.lower())

    password = self.password.get_secret_value()
    input_client_kwargs = dict(
        url=self.url, username=self.username, password=password
    )
    input_client_kwargs.update(**client_kwargs)

    if client_type == ClientType.CLOUD:
        client = Cloud(**input_client_kwargs)
    else:
        client = Bitbucket(**input_client_kwargs)
    return client

BitBucketRepository

Bases: ReadableDeploymentStorage

Interact with files stored in BitBucket repositories.

An accessible installation of git is required for this block to function properly.

Source code in prefect_bitbucket/repository.py

class BitBucketRepository(ReadableDeploymentStorage):
    """Interact with files stored in BitBucket repositories.

    An accessible installation of git is required for this block to function
    properly.
    """

    _block_type_name = "BitBucket Repository"
    _logo_url = "https://cdn.sanity.io/images/3ugk85nk/production/5d729f7355fb6828c4b605268ded9cfafab3ae4f-250x250.png"  # noqa
    _description = "Interact with files stored in BitBucket repositories."

    repository: str = Field(
        default=...,
        description="The URL of a BitBucket repository to read from in HTTPS format",
    )
    reference: Optional[str] = Field(
        default=None,
        description="An optional reference to pin to; can be a branch or tag.",
    )
    bitbucket_credentials: Optional[BitBucketCredentials] = Field(
        default=None,
        description=(
            "An optional BitBucketCredentials block for authenticating with "
            "private BitBucket repos."
        ),
    )

    @model_validator(mode="after")
    def _ensure_credentials_go_with_https(self) -> Self:
        """Ensure that credentials are not provided with 'SSH' formatted BitBucket URLs.

        Validators are by default only called on provided arguments.

        Note: validates `credentials` specifically so that it only fires when private
        repositories are used.
        """
        if self.bitbucket_credentials is not None:
            if urlparse(self.repository).scheme != "https":
                raise InvalidRepositoryURLError(
                    (
                        "Credentials can only be used with BitBucket repositories "
                        "using the 'HTTPS' format. You must either remove the "
                        "credential if you wish to use the 'SSH' format and are not "
                        "using a private repository, or you must change the repository "
                        "URL to the 'HTTPS' format."
                    )
                )

        return self

    def _create_repo_url(self) -> str:
        """Format the URL provided to the `git clone` command.

        For private repos in the cloud:
        https://x-token-auth:<access-token>@bitbucket.org/<user>/<repo>.git
        For private repos with a local bitbucket server:
        https://<username>:<access-token>@<server>/scm/<project>/<repo>.git

        All other repos should be the same as `self.repository`.
        """
        url_components = urlparse(self.repository)
        token_is_set = (
            self.bitbucket_credentials is not None and self.bitbucket_credentials.token
        )

        # Need a token for private repos
        if url_components.scheme == "https" and token_is_set:
            token = self.bitbucket_credentials.token.get_secret_value()
            username = self.bitbucket_credentials.username
            if username is None:
                username = "x-token-auth"
            updated_components = url_components._replace(
                netloc=f"{username}:{token}@{url_components.netloc}"
            )
            full_url = urlunparse(updated_components)
        else:
            full_url = self.repository

        return full_url

    @staticmethod
    def _get_paths(
        dst_dir: Union[str, None], src_dir: str, sub_directory: Optional[str]
    ) -> Tuple[str, str]:
        """Return the fully formed paths for BitBucketRepository contents.

        Return will take the form of (content_source, content_destination).

        """
        if dst_dir is None:
            content_destination = Path(".").absolute()
        else:
            content_destination = Path(dst_dir)

        content_source = Path(src_dir)

        if sub_directory:
            content_destination = content_destination.joinpath(sub_directory)
            content_source = content_source.joinpath(sub_directory)

        return str(content_source), str(content_destination)

    @sync_compatible
    async def get_directory(
        self, from_path: Optional[str] = None, local_path: Optional[str] = None
    ) -> None:
        """Clones a BitBucket project within `from_path` to the provided `local_path`.

        This defaults to cloning the repository reference configured on the
        Block to the present working directory.

        Args:
            from_path: If provided, interpreted as a subdirectory of the underlying
                repository that will be copied to the provided local path.
            local_path: A local path to clone to; defaults to present working directory.

        """
        # Construct command
        cmd = ["git", "clone", self._create_repo_url()]
        if self.reference:
            cmd += ["-b", self.reference]

        # Limit git history
        cmd += ["--depth", "1"]

        # Clone to a temporary directory and move the subdirectory over
        with TemporaryDirectory(suffix="prefect") as tmp_dir:
            cmd.append(tmp_dir)

            err_stream = io.StringIO()
            out_stream = io.StringIO()
            process = await run_process(cmd, stream_output=(out_stream, err_stream))
            if process.returncode != 0:
                err_stream.seek(0)
                raise OSError(f"Failed to pull from remote:\n {err_stream.read()}")

            content_source, content_destination = self._get_paths(
                dst_dir=local_path, src_dir=tmp_dir, sub_directory=from_path
            )

            copytree(src=content_source, dst=content_destination, dirs_exist_ok=True)

get_directory(from_path=None, local_path=None) async

Clones a BitBucket project within from_path to the provided local_path.

This defaults to cloning the repository reference configured on the Block to the present working directory.

Parameters:

Name	Type	Description	Default
from_path	Optional[str]	If provided, interpreted as a subdirectory of the underlying repository that will be copied to the provided local path.	None
local_path	Optional[str]	A local path to clone to; defaults to present working directory.	None

Source code in prefect_bitbucket/repository.py

@sync_compatible
async def get_directory(
    self, from_path: Optional[str] = None, local_path: Optional[str] = None
) -> None:
    """Clones a BitBucket project within `from_path` to the provided `local_path`.

    This defaults to cloning the repository reference configured on the
    Block to the present working directory.

    Args:
        from_path: If provided, interpreted as a subdirectory of the underlying
            repository that will be copied to the provided local path.
        local_path: A local path to clone to; defaults to present working directory.

    """
    # Construct command
    cmd = ["git", "clone", self._create_repo_url()]
    if self.reference:
        cmd += ["-b", self.reference]

    # Limit git history
    cmd += ["--depth", "1"]

    # Clone to a temporary directory and move the subdirectory over
    with TemporaryDirectory(suffix="prefect") as tmp_dir:
        cmd.append(tmp_dir)

        err_stream = io.StringIO()
        out_stream = io.StringIO()
        process = await run_process(cmd, stream_output=(out_stream, err_stream))
        if process.returncode != 0:
            err_stream.seek(0)
            raise OSError(f"Failed to pull from remote:\n {err_stream.read()}")

        content_source, content_destination = self._get_paths(
            dst_dir=local_path, src_dir=tmp_dir, sub_directory=from_path
        )

        copytree(src=content_source, dst=content_destination, dirs_exist_ok=True)