mirror of
https://github.com/aiogram/aiogram.git
synced 2026-04-08 16:37:47 +00:00
#238 Added deep linking feature
This commit is contained in:
parent
56662d6bd8
commit
a42252b5c6
3 changed files with 175 additions and 0 deletions
94
aiogram/utils/deep_linking.py
Normal file
94
aiogram/utils/deep_linking.py
Normal file
|
|
@ -0,0 +1,94 @@
|
|||
"""
|
||||
Deep linking
|
||||
|
||||
Telegram bots have a deep linking mechanism, that allows for passing additional
|
||||
parameters to the bot on startup. It could be a command that launches the bot — or
|
||||
an auth token to connect the user's Telegram account to their account on some
|
||||
external service.
|
||||
|
||||
You can read detailed description in the source:
|
||||
https://core.telegram.org/bots#deep-linking
|
||||
|
||||
We have add some utils to get deep links more handy.
|
||||
|
||||
Basic link example:
|
||||
>>> from aiogram.utils.deep_linking import get_start_link
|
||||
>>> link = await get_start_link('foo') # result: 'https://t.me/MyBot?start=foo'
|
||||
|
||||
Encoded link example:
|
||||
>>> from aiogram.utils.deep_linking import get_start_link, decode_payload
|
||||
>>> link = await get_start_link('foo', encode=True) # result: 'https://t.me/MyBot?start=Zm9v'
|
||||
>>> data = decode_payload('Zm9v') # result: 'foo'
|
||||
|
||||
"""
|
||||
|
||||
|
||||
async def get_start_link(payload: str, encode=False) -> str:
|
||||
"""
|
||||
Use this method to handy get 'start' deep link with your payload.
|
||||
If you need to encode payload or pass special characters - set encode as True
|
||||
|
||||
:param payload: args passed with /start
|
||||
:param encode: encode payload with base64url
|
||||
:return: link
|
||||
"""
|
||||
return await _create_link('start', payload, encode)
|
||||
|
||||
|
||||
async def get_startgroup_link(payload: str, encode=False) -> str:
|
||||
"""
|
||||
Use this method to handy get 'startgroup' deep link with your payload.
|
||||
If you need to encode payload or pass special characters - set encode as True
|
||||
|
||||
:param payload: args passed with /start
|
||||
:param encode: encode payload with base64url
|
||||
:return: link
|
||||
"""
|
||||
return await _create_link('startgroup', payload, encode)
|
||||
|
||||
|
||||
async def _create_link(link_type, payload: str, encode=False):
|
||||
bot = await _get_bot_user()
|
||||
payload = filter_payload(payload)
|
||||
if encode:
|
||||
payload = encode_payload(payload)
|
||||
return f'https://t.me/{bot.username}?{link_type}={payload}'
|
||||
|
||||
|
||||
def encode_payload(payload: str) -> str:
|
||||
""" Encode payload with URL-safe base64url. """
|
||||
from base64 import urlsafe_b64encode
|
||||
result: bytes = urlsafe_b64encode(payload.encode())
|
||||
return result.decode()
|
||||
|
||||
|
||||
def decode_payload(payload: str) -> str:
|
||||
""" Decode payload with URL-safe base64url. """
|
||||
from base64 import urlsafe_b64decode
|
||||
result: bytes = urlsafe_b64decode(payload + '=' * (4 - len(payload) % 4))
|
||||
return result.decode()
|
||||
|
||||
|
||||
def filter_payload(payload: str) -> str:
|
||||
""" Convert payload to text and search for not allowed symbols. """
|
||||
import re
|
||||
|
||||
# convert to string
|
||||
if not isinstance(payload, str):
|
||||
payload = str(payload)
|
||||
|
||||
# search for not allowed characters
|
||||
if re.search(r'[^_A-z0-9-]', payload):
|
||||
message = ('Wrong payload! Only A-Z, a-z, 0-9, _ and - are allowed. '
|
||||
'We recommend to encode parameters with binary and other '
|
||||
'types of content.')
|
||||
raise ValueError(message)
|
||||
|
||||
return payload
|
||||
|
||||
|
||||
async def _get_bot_user():
|
||||
""" Get current user of bot. """
|
||||
from ..bot import Bot
|
||||
bot = Bot.get_current()
|
||||
return await bot.me
|
||||
6
docs/source/utils/deep_linking.rst
Normal file
6
docs/source/utils/deep_linking.rst
Normal file
|
|
@ -0,0 +1,6 @@
|
|||
============
|
||||
Deep linking
|
||||
============
|
||||
|
||||
.. automodule:: aiogram.utils.deep_linking
|
||||
:members:
|
||||
75
tests/test_utils/test_deep_linking.py
Normal file
75
tests/test_utils/test_deep_linking.py
Normal file
|
|
@ -0,0 +1,75 @@
|
|||
import pytest
|
||||
|
||||
from aiogram.utils.deep_linking import decode_payload, encode_payload, filter_payload
|
||||
from aiogram.utils.deep_linking import get_start_link, get_startgroup_link
|
||||
from tests.types import dataset
|
||||
|
||||
# enable asyncio mode
|
||||
pytestmark = pytest.mark.asyncio
|
||||
|
||||
PAYLOADS = [
|
||||
'foo',
|
||||
'AAbbCCddEEff1122334455',
|
||||
'aaBBccDDeeFF5544332211',
|
||||
-12345678901234567890,
|
||||
12345678901234567890,
|
||||
]
|
||||
|
||||
WRONG_PAYLOADS = [
|
||||
'@BotFather',
|
||||
'spaces spaces spaces',
|
||||
1234567890123456789.0,
|
||||
]
|
||||
|
||||
|
||||
@pytest.fixture(params=PAYLOADS, name='payload')
|
||||
def payload_fixture(request):
|
||||
return request.param
|
||||
|
||||
|
||||
@pytest.fixture(params=WRONG_PAYLOADS, name='wrong_payload')
|
||||
def wrong_payload_fixture(request):
|
||||
return request.param
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def get_bot_user_fixture(monkeypatch):
|
||||
""" Monkey patching of bot.me calling. """
|
||||
from aiogram.utils import deep_linking
|
||||
|
||||
async def get_bot_user_mock():
|
||||
from aiogram.types import User
|
||||
return User(**dataset.USER)
|
||||
|
||||
monkeypatch.setattr(deep_linking, '_get_bot_user', get_bot_user_mock)
|
||||
|
||||
|
||||
class TestDeepLinking:
|
||||
async def test_get_start_link(self, payload):
|
||||
link = await get_start_link(payload)
|
||||
assert link == f'https://t.me/{dataset.USER["username"]}?start={payload}'
|
||||
|
||||
async def test_wrong_symbols(self, wrong_payload):
|
||||
with pytest.raises(ValueError):
|
||||
await get_start_link(wrong_payload)
|
||||
|
||||
async def test_get_startgroup_link(self, payload):
|
||||
link = await get_startgroup_link(payload)
|
||||
assert link == f'https://t.me/{dataset.USER["username"]}?startgroup={payload}'
|
||||
|
||||
async def test_filter_encode_and_decode(self, payload):
|
||||
_payload = filter_payload(payload)
|
||||
encoded = encode_payload(_payload)
|
||||
print(encoded)
|
||||
decoded = decode_payload(encoded)
|
||||
assert decoded == str(payload)
|
||||
|
||||
async def test_get_start_link_with_encoding(self, payload):
|
||||
# define link
|
||||
link = await get_start_link(payload, encode=True)
|
||||
|
||||
# define reference link
|
||||
payload = filter_payload(payload)
|
||||
encoded_payload = encode_payload(payload)
|
||||
|
||||
assert link == f'https://t.me/{dataset.USER["username"]}?start={encoded_payload}'
|
||||
Loading…
Add table
Add a link
Reference in a new issue