Skip to content

factory

appimage_updater.repositories.factory

Repository factory for creating appropriate repository clients.

This module provides factory functions to create the correct repository client based on URL patterns and repository types using a dynamic registry system.

get_repository_client = get_repository_client_sync module-attribute

get_repository_client_with_probing_sync = get_repository_client_sync module-attribute

detect_repository_type(url)

Detect repository type from URL without creating client.

Parameters:

Name Type Description Default
url str

Repository URL

 required

Returns:

Type Description
str

Repository type string (e.g., 'github', 'gitlab')

Raises:

Type Description
RepositoryError

If repository type cannot be determined
Source code in src/appimage_updater/repositories/factory.py 
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
def detect_repository_type(url: str) -> str:
    """Detect repository type from URL without creating client.

    Args:
        url: Repository URL

    Returns:
        Repository type string (e.g., 'github', 'gitlab')

    Raises:
        RepositoryError: If repository type cannot be determined
    """
    registry = get_repository_registry()
    domain_service = DomainKnowledgeService()

    # Try domain knowledge first
    known_handler = domain_service.get_handler_by_domain_knowledge(url)
    if known_handler:
        return known_handler.metadata.name

    # Fall back to registry detection
    handlers = registry.get_handlers_for_url(url)
    for handler in handlers:
        if handler.can_handle_url(url):
            return handler.metadata.name

    # Default fallback to github for backward compatibility
    return "github"

get_repository_client_async(url, timeout=30, user_agent=None, source_type=None, enable_probing=True, **kwargs) async

Async version of repository client factory.

Source code in src/appimage_updater/repositories/factory.py 
137
138
139
140
141
142
143
144
145
146
async def get_repository_client_async(
    url: str,
    timeout: int = 30,
    user_agent: str | None = None,
    source_type: str | None = None,
    enable_probing: bool = True,
    **kwargs: Any,
) -> RepositoryClient:
    """Async version of repository client factory."""
    return await get_repository_client_async_impl(url, timeout, user_agent, source_type, enable_probing, **kwargs)

get_repository_client_async_impl(url, timeout=30, user_agent=None, source_type=None, enable_probing=True, **kwargs) async

Get repository client with intelligent domain knowledge and optional probing.

Source code in src/appimage_updater/repositories/factory.py 
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
async def get_repository_client_async_impl(
    url: str,
    timeout: int = 30,
    user_agent: str | None = None,
    source_type: str | None = None,
    enable_probing: bool = True,
    **kwargs: Any,
) -> RepositoryClient:
    """Get repository client with intelligent domain knowledge and optional probing."""
    domain_service = DomainKnowledgeService()

    # 1. Handle explicit source type (highest priority)
    if source_type:
        return await _try_explicit_source_type(source_type, timeout, user_agent, **kwargs)

    # 2. Fast-path: Try domain knowledge first
    client = await _try_domain_knowledge(url, domain_service, timeout, user_agent, **kwargs)
    if client:
        return client

    # 3. Registry-based detection
    client = await _try_registry_handlers(url, domain_service, timeout, user_agent, **kwargs)
    if client:
        return client

    # 4. Final fallback to dynamic download
    return await _try_fallback_handler(url, timeout, user_agent, **kwargs)

get_repository_client_sync(url, timeout=30, user_agent=None, source_type=None, enable_probing=True, **kwargs)

Synchronous wrapper for get_repository_client.

Source code in src/appimage_updater/repositories/factory.py 
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
def get_repository_client_sync(
    url: str,
    timeout: int = 30,
    user_agent: str | None = None,
    source_type: str | None = None,
    enable_probing: bool = True,
    **kwargs: Any,
) -> RepositoryClient:
    """Synchronous wrapper for get_repository_client."""
    try:
        # Check if we're already in an event loop
        asyncio.get_running_loop()
        # If we are, we need to run the coroutine differently
        with concurrent.futures.ThreadPoolExecutor() as executor:
            future = executor.submit(
                asyncio.run,
                get_repository_client_async_impl(url, timeout, user_agent, source_type, enable_probing, **kwargs),
            )
            return future.result()
    except RuntimeError:
        # No running event loop, safe to use asyncio.run
        return asyncio.run(
            get_repository_client_async_impl(url, timeout, user_agent, source_type, enable_probing, **kwargs)
        )