Webhook¶
Tip
To use webhooks you need to enable them in the @CryptoBot settings as follows:
🏝 Crypto Pay -> My Apps -> YOUR APP -> Webhooks -> 🌕 Enable webhooks.
Usage example with aiohttp web server¶
import asyncio
from aiohttp.web import Application, _run_app
from aiosend import CryptoPay
from aiosend.types import Invoice
from aiosend.webhook import AiohttpManager
app = Application()
cp = CryptoPay(
"TOKEN",
webhook_manager=AiohttpManager(app, "/handler"),
)
@cp.invoice_paid()
async def handler(invoice: Invoice) -> None:
print(f"Received {invoice.amount} {invoice.asset}")
async def main() -> None:
invoice = await cp.create_invoice(1, "USDT")
print("invoice link:", invoice.bot_invoice_url)
await _run_app(app)
if __name__ == "__main__":
asyncio.run(main())
Usage example with FastAPI web server¶
Tip
In order to use aiosend with fastapi you need to install extra package
pip install aiosend[fastapi]
Note
aiosend.webhook.FastAPIManager supports FastAPI dependency injection
in event handlers using fastapi.Depends.
See FastAPI dependencies for more information.
import asyncio
import uvicorn
from fastapi import FastAPI
from aiosend import CryptoPay
from aiosend.types import Invoice
from aiosend.webhook import FastAPIManager
app = FastAPI()
cp = CryptoPay(
"TOKEN",
webhook_manager=FastAPIManager(app, "/handler"),
)
@cp.invoice_paid()
async def handler(invoice: Invoice) -> None:
print(f"Received {invoice.amount} {invoice.asset}")
async def main() -> None:
invoice = await cp.create_invoice(1, "USDT")
print("invoice link:", invoice.bot_invoice_url)
if __name__ == "__main__":
asyncio.run(main())
uvicorn.run(app)
Usage example with Flask web server¶
Tip
In order to use aiosend with flask you need to install extra package
pip install aiosend[flask]
import asyncio
from flask import Flask
from aiosend import CryptoPay
from aiosend.types import Invoice
from aiosend.webhook import FlaskManager
app = Flask(__name__)
cp = CryptoPay(
"TOKEN",
webhook_manager=FlaskManager(app, "/handler"),
)
@cp.invoice_paid()
async def handler(invoice: Invoice) -> None:
print(f"Received {invoice.amount} {invoice.asset}")
async def main() -> None:
invoice = await cp.create_invoice(1, "USDT")
print("invoice link:", invoice.bot_invoice_url)
if __name__ == "__main__":
asyncio.run(main())
app.run()
Usage example with Litestar web server¶
Tip
In order to use aiosend with Litestar you need to install extra package
pip install aiosend[litestar]
import asyncio
import uvicorn
from litestar import Litestar
from aiosend import CryptoPay
from aiosend.types import Invoice
from aiosend.webhook import LitestarManager
app = Litestar()
cp = CryptoPay(
"TOKEN",
webhook_manager=LitestarManager(app, "/handler"),
)
@cp.invoice_paid()
async def handler(invoice: Invoice) -> None:
print(f"Received {invoice.amount} {invoice.asset}")
async def main() -> None:
invoice = await cp.create_invoice(1, "USDT")
print("invoice link:", invoice.bot_invoice_url)
if __name__ == "__main__":
asyncio.run(main())
uvicorn.run(app)
Usage example with Starlette web server¶
Tip
In order to use aiosend with fastapi you need to install extra package
pip install aiosend[starlette]
import asyncio
import uvicorn
from starlette.applications import Starlette
from aiosend import CryptoPay
from aiosend.types import Invoice
from aiosend.webhook import StarletteManager
app = Starlette()
cp = CryptoPay(
"TOKEN",
webhook_manager=StarletteManager(app, "/handler"),
)
@cp.invoice_paid()
async def handler(invoice: Invoice) -> None:
print(f"Received {invoice.amount} {invoice.asset}")
async def main() -> None:
invoice = await cp.create_invoice(1, "USDT")
print("invoice link:", invoice.bot_invoice_url)
if __name__ == "__main__":
asyncio.run(main())
uvicorn.run(app)
aiosend uses aiohttp as web server by default.
You can implement your own webhook manager by inheriting aiosend.webhook.WebhookManager
and overriding aiosend.webhook.WebhookManager.register_handler.
- class aiosend.webhook.WebhookManager(app, path)¶
Webhook manager.
If you want to implement your own webhook manager, you must inherit from this class.
- abstractmethod register_handler(feed_update)¶
Register webhook handler.
Override this method in your own webhook manager class. This method is used for registering webhook handler in your app.
- Parameters:
feed_update (
Callable[[Concatenate[bytes,Mapping[str,str],ParamSpec(P, bound=None)]],Awaitable[bool]]) – Web server handler object.- Return type:
None
- class aiosend.webhook.AiohttpManager(app, path)¶
Bases:
WebhookManager[Application]aiohttp webhook manager.
Webhook manager based on aiohttp.
- register_handler(feed_update)¶
Register webhook handler.
- Return type:
None
- class aiosend.webhook.FastAPIManager(app, path, dependencies=None)¶
Bases:
WebhookManager[FastAPI | APIRouter]FastAPI webhook manager.
Webhook manager based on FastAPI.
- register_handler(feed_update)¶
Register webhook handler.
- Return type:
None
- class aiosend.webhook.FlaskManager(app, path)¶
Bases:
WebhookManager[Flask]Flask webhook manager.
Webhook manager based on Flask.
- register_handler(feed_update)¶
Register webhook handler.
- Return type:
None
- class aiosend.webhook.LitestarManager(app, path)¶
Bases:
WebhookManager[Litestar | Router]Litestar webhook manager.
` Webhook manager based on `Litestar <https://litestar.dev/>`_. `- register_handler(feed_update)¶
Register webhook handler.
- Return type:
None
- class aiosend.webhook.StarletteManager(app, path)¶
Bases:
WebhookManager[Starlette | Router]Starlette webhook manager.
Webhook manager based on Starlette.
- register_handler(feed_update)¶
Register webhook handler.
- Return type:
None