`vendor_fabric.github`¶

GitHub connector using shared extended-data primitives.

Submodules¶

vendor_fabric.github.tools

Package Contents¶

Classes¶

GitHubConnector

GitHub connector for repository operations.

Functions¶

`get_github_api_error`	Extract error message from a GitHub exception.
`build_github_actions_workflow`	Generate a GitHub Actions workflow YAML string.

Data¶

`Auth`
`Github`
`GraphqlClient`
`GithubException`
`UnknownObjectException`
`FilePath`
`DEFAULT_PER_PAGE`

API¶

vendor_fabric.github.Auth: Any = None¶

vendor_fabric.github.Github: Any = None¶

vendor_fabric.github.GraphqlClient: Any = None¶

exception vendor_fabric.github.GitHubFallbackError¶

Bases: Exception

Fallback exception used until PyGithub is imported.

Initialization

Initialize self. See help(type(self)) for accurate signature.

add_note()¶

class args¶

with_traceback()¶

vendor_fabric.github.GithubException: Any = None¶

vendor_fabric.github.UnknownObjectException: Any = None¶

vendor_fabric.github.FilePath = None¶

vendor_fabric.github.get_github_api_error(exc: BaseException) → str | None¶: Extract error message from a GitHub exception.

vendor_fabric.github.DEFAULT_PER_PAGE = 100¶

class vendor_fabric.github.GitHubConnector(github_owner: str, github_repo: str | None = None, github_branch: str | None = None, github_token: str | None = None, per_page: int = DEFAULT_PER_PAGE, logger: extended_data.logging.Logging | None = None, **kwargs: Any)¶

Bases: vendor_fabric.base.ConnectorBase

GitHub connector for repository operations.

Initialization

Initialize the connector.

Args: api_key: API key (overrides environment variable) base_url: Base URL (overrides class default) timeout: HTTP timeout in seconds logger: Logger instance **kwargs: Passed to InputProvider

get_repository_branch(branch_name: str) → Any | None¶: Get a repository branch by name.

create_repository_branch(branch_name: str, parent_branch: str | None = None) → Any | None¶: Create a new repository branch.

get_repository_file(file_path: vendor_fabric.github.FilePath, decode: bool | None = True, return_sha: bool | None = False, return_path: bool | None = False, charset: str | None = 'utf-8', errors: str | None = 'strict', raise_on_not_found: bool = False) → extended_data.containers.ExtendedDict | extended_data.containers.ExtendedList[Any] | extended_data.containers.ExtendedString | extended_data.containers.ExtendedTuple[Any] | None¶: Get a file from the repository.

update_repository_file(file_path: vendor_fabric.github.FilePath, file_data: Any, file_sha: str | None = None, msg: str | None = None, allow_encoding: bool | str | None = None, allow_empty: bool = False, **format_opts: Any) → Any | None¶: Update a file in the repository.

delete_repository_file(file_path: vendor_fabric.github.FilePath, msg: str | None = None) → Any | None¶: Delete a file from the repository.

list_org_members(role: str | None = None, include_pending: bool = False) → extended_data.containers.ExtendedDict¶

List organization members.

Args: role: Filter by role (‘admin’, ‘member’). None returns all. include_pending: Include pending invitations. Defaults to False.

Returns: Dictionary mapping usernames to member data.

get_org_member(username: str) → extended_data.containers.ExtendedDict | None¶

Get a specific organization member.

Args: username: GitHub username.

Returns: Member data or None if not found.

list_repositories(type_filter: str = 'all', include_branches: bool = False) → extended_data.containers.ExtendedDict¶

List organization repositories.

Args: type_filter: Filter type (‘all’, ‘public’, ‘private’, ‘forks’, ‘sources’, ‘member’). include_branches: Include branch information. Defaults to False.

Returns: Dictionary mapping repo names to repository data.

get_repository(repo_name: str) → extended_data.containers.ExtendedDict | None¶

Get a specific repository.

Args: repo_name: Repository name.

Returns: Repository data or None if not found.

list_teams(include_members: bool = False, include_repos: bool = False) → extended_data.containers.ExtendedDict¶

List organization teams.

Args: include_members: Include team members. Defaults to False. include_repos: Include team repositories. Defaults to False.

Returns: Dictionary mapping team slugs to team data.

get_team(team_slug: str) → extended_data.containers.ExtendedDict | None¶

Get a specific team.

Args: team_slug: Team slug.

Returns: Team data or None if not found.

add_team_member(team_slug: str, username: str, role: str = 'member') → bool¶

Add a member to a team.

Args: team_slug: Team slug. username: GitHub username. role: Role (‘member’ or ‘maintainer’). Defaults to ‘member’.

Returns: True if successful.

remove_team_member(team_slug: str, username: str) → bool¶

Remove a member from a team.

Args: team_slug: Team slug. username: GitHub username.

Returns: True if successful.

execute_graphql(query: str, variables: dict[str, Any] | None = None) → extended_data.containers.ExtendedDict¶

Execute a GraphQL query against the GitHub API.

Args: query: GraphQL query string. variables: Optional query variables.

Returns: Query response data.

get_users_with_verified_emails(members: collections.abc.Mapping[str, collections.abc.Mapping[str, Any]] | None = None, domain_filter: str | None = None) → extended_data.containers.ExtendedDict¶

Get organization members with their verified emails.

Uses GraphQL to get verified email addresses for org members.

Args: members: Pre-fetched members dict. Fetched if not provided. domain_filter: Filter by email domain (e.g., ‘company.com’).

Returns: Dictionary mapping usernames to member data with verified emails.

build_workflow(name: str, on: collections.abc.Mapping[str, Any], jobs: collections.abc.Mapping[str, collections.abc.Mapping[str, Any]], env: collections.abc.Mapping[str, str] | None = None, permissions: collections.abc.Mapping[str, str] | None = None, concurrency: collections.abc.Mapping[str, Any] | None = None, defaults: collections.abc.Mapping[str, Any] | None = None) → extended_data.containers.ExtendedDict¶

Build a GitHub Actions workflow structure.

Args: name: Workflow name. on: Trigger configuration. jobs: Jobs configuration. env: Global environment variables. permissions: Workflow permissions. concurrency: Concurrency settings. defaults: Default settings for jobs.

Returns: Workflow configuration dict suitable for YAML export.

build_workflow_job(runs_on: str = 'ubuntu-latest', steps: collections.abc.Sequence[collections.abc.Mapping[str, Any]] | None = None, needs: collections.abc.Sequence[str] | None = None, if_condition: str | None = None, env: collections.abc.Mapping[str, str] | None = None, strategy: collections.abc.Mapping[str, Any] | None = None, timeout_minutes: int | None = None, services: collections.abc.Mapping[str, Any] | None = None, outputs: collections.abc.Mapping[str, str] | None = None) → extended_data.containers.ExtendedDict¶

Build a GitHub Actions workflow job.

Args: runs_on: Runner label(s). steps: Job steps. needs: Dependencies on other jobs. if_condition: Conditional expression. env: Job environment variables. strategy: Matrix strategy. timeout_minutes: Job timeout. services: Service containers. outputs: Job outputs.

Returns: Job configuration dict.

Build a GitHub Actions workflow step.

Args: name: Step name. uses: Action to use (e.g., ‘actions/checkout@v4’). run: Shell command(s) to run. with_params: Input parameters for the action. env: Step environment variables. if_condition: Conditional expression. working_directory: Working directory for run commands. shell: Shell to use for run commands. id: Step ID for outputs.

Returns: Step configuration dict.

create_python_ci_workflow(python_versions: list[str] | None = None, test_command: str = 'pytest', lint_command: str = 'ruff check', format_command: str | None = 'ruff format --check', install_command: str = 'uv sync --all-packages', working_directory: str = '.') → extended_data.containers.ExtendedDict¶

Create a standard Python CI workflow.

Args: python_versions: Python versions to test. Defaults to [‘3.12’]. test_command: Test command. Defaults to ‘pytest’. lint_command: Lint command. Defaults to ‘ruff check’. format_command: Format check command. None to skip. install_command: Dependency install command. working_directory: Working directory for commands.

Returns: Complete workflow configuration.

BASE_URL: ClassVar[str] = <Multiline-String>¶

API_KEY_ENV: ClassVar[str] = <Multiline-String>¶

CONNECTOR_CATEGORY: ClassVar[str] = 'external'¶

CONNECTOR_CAPABILITIES: ClassVar[tuple[str, ...]] = ()¶

TIMEOUT: ClassVar[float] = 300.0¶

MIN_REQUEST_INTERVAL: ClassVar[float] = 0.0¶

MAX_RETRIES: ClassVar[int] = 5¶

property api_key: str¶

property client: httpx.Client¶

close() → None¶

request(method: str, endpoint: str, *, headers: dict[str, str] | None = None, **kwargs: Any) → httpx.Response¶

decode_response(response: httpx.Response, *, suffix: str | None = None, as_extended: bool = True) → Any¶

decode_response_file(response: httpx.Response, *, source: str | None = None, suffix: str | None = None, as_extended: bool = True, metadata: collections.abc.Mapping[str, Any] | None = None) → extended_data.io.DataFile¶

extend_result(value: Any) → Any¶

request_data(method: str, endpoint: str, *, headers: dict[str, str] | None = None, suffix: str | None = None, as_extended: bool = True, **kwargs: Any) → Any¶

request_data_file(method: str, endpoint: str, *, headers: dict[str, str] | None = None, suffix: str | None = None, as_extended: bool = True, **kwargs: Any) → extended_data.io.DataFile¶

request_workflow(method: str, endpoint: str, *, headers: dict[str, str] | None = None, suffix: str | None = None, as_extended: bool = True, **kwargs: Any) → extended_data.workflows.DataWorkflow¶

get(endpoint: str, **kwargs: Any) → httpx.Response¶

get_data(endpoint: str, *, suffix: str | None = None, as_extended: bool = True, **kwargs: Any) → Any¶

get_workflow(endpoint: str, *, suffix: str | None = None, as_extended: bool = True, **kwargs: Any) → extended_data.workflows.DataWorkflow¶

post(endpoint: str, **kwargs: Any) → httpx.Response¶

post_data(endpoint: str, *, suffix: str | None = None, as_extended: bool = True, **kwargs: Any) → Any¶

post_workflow(endpoint: str, *, suffix: str | None = None, as_extended: bool = True, **kwargs: Any) → extended_data.workflows.DataWorkflow¶

put(endpoint: str, **kwargs: Any) → httpx.Response¶

put_data(endpoint: str, *, suffix: str | None = None, as_extended: bool = True, **kwargs: Any) → Any¶

put_workflow(endpoint: str, *, suffix: str | None = None, as_extended: bool = True, **kwargs: Any) → extended_data.workflows.DataWorkflow¶

delete(endpoint: str, **kwargs: Any) → httpx.Response¶

delete_data(endpoint: str, *, suffix: str | None = None, as_extended: bool = True, **kwargs: Any) → Any¶

delete_workflow(endpoint: str, *, suffix: str | None = None, as_extended: bool = True, **kwargs: Any) → extended_data.workflows.DataWorkflow¶

patch(endpoint: str, **kwargs: Any) → httpx.Response¶

patch_data(endpoint: str, *, suffix: str | None = None, as_extended: bool = True, **kwargs: Any) → Any¶

patch_workflow(endpoint: str, *, suffix: str | None = None, as_extended: bool = True, **kwargs: Any) → extended_data.workflows.DataWorkflow¶

download(url: str, output_path: str) → int¶

vendor_capabilities: ClassVar[dict[str, vendor_fabric.capabilities.CapabilitySpec]] = None¶

vendor_capability_methods: ClassVar[dict[str, str]] = None¶

get_input(k: str, default: Any | None = None, required: bool = False, is_bool: bool = False, is_integer: bool = False, is_float: bool = False, is_path: bool = False, is_datetime: bool = False, as_extended: bool = False) → Any¶

decode_input(k: str, default: Any | None = None, required: bool = False, decode_from_json: bool = False, decode_from_yaml: bool = False, decode_from_base64: bool = False, allow_none: bool = True, as_extended: bool = False) → Any¶

freeze_inputs() → extended_data.containers.mappings.ExtendedDict¶

thaw_inputs() → extended_data.containers.mappings.ExtendedDict¶

snapshot_inputs(*, frozen: bool = False) → extended_data.containers.mappings.ExtendedDict¶

replace_inputs(new_inputs: collections.abc.Mapping[str, Any] | None, *, clear_frozen: bool = True) → extended_data.containers.mappings.ExtendedDict¶

merge_inputs(new_inputs: collections.abc.Mapping[str, Any] | None) → extended_data.containers.mappings.ExtendedDict¶

shift_inputs() → extended_data.containers.mappings.ExtendedDict¶

vendor_fabric.github.build_github_actions_workflow(workflow_name: str, jobs: dict[str, Any], concurrency_group: str | None = None, environment_variables: dict[str, str] | None = None, secrets: dict[str, str] | None = None, use_oidc_auth: bool = True, events: dict[str, Any] | None = None, triggers: dict[str, Any] | None = None, inputs: dict[str, Any] | None = None, pull_requests: dict[str, Any] | None = None) → str¶: Generate a GitHub Actions workflow YAML string.

vendor_fabric.github¶