Skip to content

BoostyClient

boostylib.client.BoostyClient

Main client for the Boosty.to API.

Provides access to all API modules, event system, and caching.

Example::

async with BoostyClient(access_token="...") as client:
    user = await client.users.get_current_user()
    print(user.name)

Parameters:

Name Type Description Default
access_token str | None

Static access token (no auto-refresh).

 None
credentials AuthCredentials | None

Full credentials for auto-refresh.

 None
token_storage TokenStorage | None

Custom token storage backend.

 None
settings BoostySettings | None

Client configuration.

 None
http_client AsyncClient | None

Custom httpx.AsyncClient (for testing or proxies).

 None
middleware list[Any] | None

List of middleware instances.

 None
cache CacheBackend | None

Custom cache backend (default: MemoryCache if enabled).

 None
blog_username str

Blog username for event polling.

 ''
Source code in src/boostylib/client.py 
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 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
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
class BoostyClient:
    """Main client for the Boosty.to API.

    Provides access to all API modules, event system, and caching.

    Example::

        async with BoostyClient(access_token="...") as client:
            user = await client.users.get_current_user()
            print(user.name)

    Args:
        access_token: Static access token (no auto-refresh).
        credentials: Full credentials for auto-refresh.
        token_storage: Custom token storage backend.
        settings: Client configuration.
        http_client: Custom httpx.AsyncClient (for testing or proxies).
        middleware: List of middleware instances.
        cache: Custom cache backend (default: MemoryCache if enabled).
        blog_username: Blog username for event polling.
    """

    def __init__(
        self,
        *,
        access_token: str | None = None,
        credentials: AuthCredentials | None = None,
        token_storage: TokenStorage | None = None,
        settings: BoostySettings | None = None,
        http_client: httpx.AsyncClient | None = None,
        middleware: list[Any] | None = None,
        cache: CacheBackend | None = None,
        blog_username: str = "",
    ) -> None:
        self._settings = settings or BoostySettings()
        self._middleware_chain = MiddlewareChain(middleware)

        # Auth
        self._auth_manager = AuthManager(
            settings=self._settings,
            token_storage=token_storage,
            credentials=credentials,
            access_token=access_token,
        )

        # Transport
        self._transport = HTTPTransport(
            settings=self._settings,
            auth_manager=self._auth_manager,
            http_client=http_client,
            middleware_chain=self._middleware_chain,
        )

        # Cache
        self.cache = CacheManager(
            backend=cache,
            enabled=self._settings.cache_enabled,
            ttl_blog=self._settings.cache_ttl_blog,
            ttl_levels=self._settings.cache_ttl_levels,
            ttl_user=self._settings.cache_ttl_user,
            ttl_default=self._settings.cache_ttl_default,
        )

        # API modules
        self.users = UsersAPI(self._transport)
        self.blogs = BlogsAPI(self._transport)
        self.posts = PostsAPI(self._transport)
        self.comments = CommentsAPI(self._transport)
        self.subscriptions = SubscriptionsAPI(self._transport)
        self.donations = DonationsAPI(self._transport)
        self.targets = TargetsAPI(self._transport)
        self.showcase = ShowcaseAPI(self._transport)
        self.media = MediaAPI(self._transport)
        self.bundles = BundlesAPI(self._transport)

        # Events
        self._dispatcher = EventDispatcher()
        self._blog_username = blog_username
        self._poller = EventPoller(
            settings=self._settings,
            dispatcher=self._dispatcher,
            blog_username=blog_username,
            posts_api=self.posts,
            subscriptions_api=self.subscriptions,
            comments_api=self.comments,
        )

    def on(self, event_type: EventType) -> Callable[[EventHandler], EventHandler]:
        """Decorator to register an event handler.

        Example::

            @client.on(EventType.NEW_DONATION)
            async def handle(event):
                print(event.amount)
        """
        return self._dispatcher.on(event_type)

    async def start_polling(self, blog_username: str | None = None) -> None:
        """Start the event polling loop. Blocks until stopped.

        Args:
            blog_username: Override the blog to poll (if not set in constructor).
        """
        if blog_username:
            self._poller._blog = blog_username
        await self._poller.start()
        try:
            while self._poller._running:
                await asyncio.sleep(1)
        except asyncio.CancelledError:
            await self._poller.stop()

    async def stop_polling(self) -> None:
        """Stop the event polling loop."""
        await self._poller.stop()

    async def close(self) -> None:
        """Close the client and release resources."""
        await self._poller.stop()
        await self._transport.close()

    async def __aenter__(self) -> Self:
        await self._auth_manager.initialize()
        return self

    async def __aexit__(self, *args: Any) -> None:
        await self.close()

users = UsersAPI(self._transport) instance-attribute

blogs = BlogsAPI(self._transport) instance-attribute

posts = PostsAPI(self._transport) instance-attribute

comments = CommentsAPI(self._transport) instance-attribute

subscriptions = SubscriptionsAPI(self._transport) instance-attribute

donations = DonationsAPI(self._transport) instance-attribute

targets = TargetsAPI(self._transport) instance-attribute

showcase = ShowcaseAPI(self._transport) instance-attribute

media = MediaAPI(self._transport) instance-attribute

cache = CacheManager(backend=cache, enabled=(self._settings.cache_enabled), ttl_blog=(self._settings.cache_ttl_blog), ttl_levels=(self._settings.cache_ttl_levels), ttl_user=(self._settings.cache_ttl_user), ttl_default=(self._settings.cache_ttl_default)) instance-attribute

bundles = BundlesAPI(self._transport) instance-attribute

__init__(*, access_token=None, credentials=None, token_storage=None, settings=None, http_client=None, middleware=None, cache=None, blog_username='')

Source code in src/boostylib/client.py 
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 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
109
110
111
112
113
114
115
116
117
118
def __init__(
    self,
    *,
    access_token: str | None = None,
    credentials: AuthCredentials | None = None,
    token_storage: TokenStorage | None = None,
    settings: BoostySettings | None = None,
    http_client: httpx.AsyncClient | None = None,
    middleware: list[Any] | None = None,
    cache: CacheBackend | None = None,
    blog_username: str = "",
) -> None:
    self._settings = settings or BoostySettings()
    self._middleware_chain = MiddlewareChain(middleware)

    # Auth
    self._auth_manager = AuthManager(
        settings=self._settings,
        token_storage=token_storage,
        credentials=credentials,
        access_token=access_token,
    )

    # Transport
    self._transport = HTTPTransport(
        settings=self._settings,
        auth_manager=self._auth_manager,
        http_client=http_client,
        middleware_chain=self._middleware_chain,
    )

    # Cache
    self.cache = CacheManager(
        backend=cache,
        enabled=self._settings.cache_enabled,
        ttl_blog=self._settings.cache_ttl_blog,
        ttl_levels=self._settings.cache_ttl_levels,
        ttl_user=self._settings.cache_ttl_user,
        ttl_default=self._settings.cache_ttl_default,
    )

    # API modules
    self.users = UsersAPI(self._transport)
    self.blogs = BlogsAPI(self._transport)
    self.posts = PostsAPI(self._transport)
    self.comments = CommentsAPI(self._transport)
    self.subscriptions = SubscriptionsAPI(self._transport)
    self.donations = DonationsAPI(self._transport)
    self.targets = TargetsAPI(self._transport)
    self.showcase = ShowcaseAPI(self._transport)
    self.media = MediaAPI(self._transport)
    self.bundles = BundlesAPI(self._transport)

    # Events
    self._dispatcher = EventDispatcher()
    self._blog_username = blog_username
    self._poller = EventPoller(
        settings=self._settings,
        dispatcher=self._dispatcher,
        blog_username=blog_username,
        posts_api=self.posts,
        subscriptions_api=self.subscriptions,
        comments_api=self.comments,
    )

on(event_type)

Decorator to register an event handler.

Example::

@client.on(EventType.NEW_DONATION)
async def handle(event):
    print(event.amount)
Source code in src/boostylib/client.py 
120
121
122
123
124
125
126
127
128
129
def on(self, event_type: EventType) -> Callable[[EventHandler], EventHandler]:
    """Decorator to register an event handler.

    Example::

        @client.on(EventType.NEW_DONATION)
        async def handle(event):
            print(event.amount)
    """
    return self._dispatcher.on(event_type)

start_polling(blog_username=None) async

Start the event polling loop. Blocks until stopped.

Parameters:

Name Type Description Default
blog_username str | None

Override the blog to poll (if not set in constructor).

 None
Source code in src/boostylib/client.py 
131
132
133
134
135
136
137
138
139
140
141
142
143
144
async def start_polling(self, blog_username: str | None = None) -> None:
    """Start the event polling loop. Blocks until stopped.

    Args:
        blog_username: Override the blog to poll (if not set in constructor).
    """
    if blog_username:
        self._poller._blog = blog_username
    await self._poller.start()
    try:
        while self._poller._running:
            await asyncio.sleep(1)
    except asyncio.CancelledError:
        await self._poller.stop()

stop_polling() async

Stop the event polling loop.

Source code in src/boostylib/client.py 
146
147
148
async def stop_polling(self) -> None:
    """Stop the event polling loop."""
    await self._poller.stop()

close() async

Close the client and release resources.

Source code in src/boostylib/client.py 
150
151
152
153
async def close(self) -> None:
    """Close the client and release resources."""
    await self._poller.stop()
    await self._transport.close()