End-to-End Examples¶

import os
from contextlib import asynccontextmanager

from fastapi import FastAPI, HTTPException
from v2hub import AsyncVPNClient, ConflictError, VPNAPIError


@asynccontextmanager
async def lifespan(app: FastAPI):
    app.state.v2hub = AsyncVPNClient(
        base_url=os.environ["V2HUB_API_URL"],
        api_token=os.environ["V2HUB_API_TOKEN"],
    )
    await app.state.v2hub.connect()
    yield
    await app.state.v2hub.close()


app = FastAPI(lifespan=lifespan)


@app.post("/signup/{username}/provision-vpn")
async def provision_vpn(username: str):
    client = app.state.v2hub
    sub_name = f"user-{username}-vpn"

    try:
        sub = await client.create_subscription(sub_name)
    except ConflictError:
        # Already provisioned (e.g. a retried request) — look it up instead
        # of failing the signup.
        subs = await client.list_subscriptions()
        sub = next((s for s in subs if s.name == sub_name), None)
        if sub is None:
            raise HTTPException(500, "subscription conflict but not found")
    except VPNAPIError as e:
        raise HTTPException(502, f"Could not provision VPN: {e.recovery_hint}")

    return {"subscription_url": f"{os.environ['V2HUB_API_URL']}/sub/{sub.token}"}

import asyncio
from v2hub import AsyncVPNClient, NotFoundError, VPNAPIError

PROVIDER_TOKEN = "provider-api-token"
BASE_URL = "https://api.example.com"


async def onboard_user(client: AsyncVPNClient, telegram_user_id: int, v2hub_user_id: int):
    """Runs when a Telegram user first interacts with the bot."""
    await client.create_provider_connection(user_id=v2hub_user_id)

    sub = await client.create_subscription(
        f"tg-{telegram_user_id}",
        as_provider_for_user_id=v2hub_user_id,
    )
    return sub.token


async def add_server_for_user(client: AsyncVPNClient, v2hub_user_id: int, sub_token: str, config_uri: str):
    """Runs when the user sends the bot a new config string to add."""
    try:
        await client.add_sources(sub_token, [config_uri], as_provider_for_user_id=v2hub_user_id)
    except NotFoundError:
        # Subscription token the bot had stored is stale — treat as a bug
        # in the bot's own storage, not a user-facing retry case.
        raise RuntimeError(f"stored token {sub_token} no longer exists")
    except VPNAPIError as e:
        return f"Couldn't add that server: {e.recovery_hint}"
    return "Added!"


async def main():
    async with AsyncVPNClient(BASE_URL, PROVIDER_TOKEN) as client:
        token = await onboard_user(client, telegram_user_id=555111, v2hub_user_id=98765)
        print(f"Onboarded. Subscription URL: {BASE_URL}/sub/{token}")

        result = await add_server_for_user(
            client, v2hub_user_id=98765, sub_token=token,
            config_uri="vless://uuid@server1:443#Server1",
        )
        print(result)


asyncio.run(main())

import asyncio
import logging
import os

from v2hub import VPNAPIError
from v2hub_admin import AsyncAdminClient

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("nightly-report")


async def main():
    async with AsyncAdminClient(
        base_url=os.environ["V2HUB_API_URL"],
        secret_key=os.environ["V2HUB_ADMIN_SECRET"],
    ) as admin:
        try:
            stats = await admin.get_stats(period="week")
        except VPNAPIError as e:
            logger.error("failed to fetch weekly stats: %s", e)
            return

        logger.info("weekly stats: %s", stats)

        bans = await admin.get_ban_list()
        logger.info("%d active ban(s)", bans.total)
        for ban in bans.entries:
            status = await admin.get_ban_status(ban.ip_address)
            if not status.is_banned:
                # Ban expired between listing and checking — nothing to do,
                # but useful to note for auditing.
                logger.info("%s ban already expired", ban.ip_address)


asyncio.run(main())