Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_parse_mode(self):
query = AnswerInlineQuery(
inline_query_id="query id",
results=[InlineQueryResultPhoto(id="result id", photo_url="photo", thumb_url="thumb")],
)
request = query.build_request()
assert request.data["results"][0]["parse_mode"] is None
token = Bot.set_current(Bot(token="42:TEST", parse_mode="HTML"))
try:
request = query.build_request()
assert request.data["results"][0]["parse_mode"] == "HTML"
finally:
Bot.reset_current(token)
async def bot(event_loop):
""" Bot fixture """
_bot = Bot(token='123456789:AABBCCDDEEFFaabbccddeeff-1234567890',
loop=event_loop)
yield _bot
await _bot.close()
Step 5.4: compile mo files
command from step 4
"""
from pathlib import Path
from aiogram import Bot, Dispatcher, executor, types
from aiogram.contrib.middlewares.i18n import I18nMiddleware
TOKEN = "BOT TOKEN HERE"
I18N_DOMAIN = "mybot"
BASE_DIR = Path(__file__).parent
LOCALES_DIR = BASE_DIR / "locales"
bot = Bot(TOKEN, parse_mode=types.ParseMode.HTML)
dp = Dispatcher(bot)
# Setup i18n middleware
i18n = I18nMiddleware(I18N_DOMAIN, LOCALES_DIR)
dp.middleware.setup(i18n)
# Alias for gettext method
_ = i18n.gettext
@dp.message_handler(commands=["start"])
async def cmd_start(message: types.Message):
# Simply use `_('message')` instead of `'message'` and never use f-strings for translatable texts.
await message.reply(_("Hello, <b>{user}</b>!").format(user=message.from_user.full_name))
TOKEN = config['RASTREIOBOT']['TOKEN']
int_check = int(config['RASTREIOBOT']['int_check'])
LOG_ALERTS_FILE = config['RASTREIOBOT']['alerts_log']
PATREON = config['RASTREIOBOT']['patreon']
INTERVAL = 0.03
logger_info = logging.getLogger('InfoLogger')
handler_info = logging.handlers.TimedRotatingFileHandler(
LOG_ALERTS_FILE, when='midnight', interval=1, backupCount=7, encoding='utf-8'
)
#handler_info = logging.FileHandler(LOG_ALERTS_FILE)
logger_info.setLevel(logging.DEBUG)
logger_info.addHandler(handler_info)
bot = Bot(token=TOKEN)
dp = Dispatcher(bot)
client = motor.motor_asyncio.AsyncIOMotorClient()
db = client.rastreiobot
async def get_package(code):
stat = await async_check_update(code)
if stat == 0:
stat = 'Sistema dos Correios fora do ar.'
elif stat == 1:
stat = None
elif stat == 3:
stat = None
else:
cursor = await db.rastreiobot.update_one (
{ "code" : code.upper() },
{
import asyncio
from aiogram import Bot
from aiogram import types
from aiogram.dispatcher import Dispatcher
from aiogram.types.message import ContentTypes
from aiogram.utils import executor
BOT_TOKEN = "BOT TOKEN HERE"
PAYMENTS_PROVIDER_TOKEN = "123456789:TEST:1234567890abcdef1234567890abcdef"
loop = asyncio.get_event_loop()
bot = Bot(BOT_TOKEN)
dp = Dispatcher(bot, loop=loop)
# Setup prices
prices = [
types.LabeledPrice(label="Working Time Machine", amount=5750),
types.LabeledPrice(label="Gift wrapping", amount=500),
]
# Setup shipping options
shipping_options = [
types.ShippingOption(id="instant", title="WorldWide Teleporter").add(
types.LabeledPrice("Teleporter", 1000)
),
types.ShippingOption(id="pickup", title="Local pickup").add(types.LabeledPrice("Pickup", 300)),
]
import logging
from aiogram import Bot, Dispatcher, types, executor
API_TOKEN = 'API_TOKEN_HERE'
logging.basicConfig(level=logging.DEBUG)
bot = Bot(token=API_TOKEN)
dp = Dispatcher(bot=bot)
# checks specified chat
@dp.message_handler(is_chat_admin=-1001241113577)
async def handle_specified(msg: types.Message):
await msg.answer("You are an admin of the specified chat!")
# checks multiple chats
@dp.message_handler(is_chat_admin=[-1001241113577, -320463906])
async def handle_multiple(msg: types.Message):
await msg.answer("You are an admin of multiple chats!")
# checks current chat
"""
import logging
from aiogram import Bot, Dispatcher, executor, types
API_TOKEN = 'BOT_TOKEN_HERE'
# Configure logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
logger.setLevel(logging.DEBUG)
# Initialize bot and dispatcher
bot = Bot(token=API_TOKEN)
dp = Dispatcher(bot)
@dp.message_handler(commands='start')
async def start_cmd_handler(message: types.Message):
keyboard_markup = types.ReplyKeyboardMarkup(row_width=3)
# default row_width is 3, so here we can omit it actually
# kept for clearness
btns_text = ('Yes!', 'No!')
keyboard_markup.row(*(types.KeyboardButton(text) for text in btns_text))
# adds buttons as a new row to the existing keyboard
# the behaviour doesn't depend on row_width attribute
more_btns_text = (
"I don't know",
import asyncio
import logging
from aiogram import Bot, types
from aiogram.dispatcher import Dispatcher
from aiogram.types import ContentType
API_TOKEN = TOKEN = 'BOT TOKEN HERE'
logging.basicConfig(level=logging.INFO)
loop = asyncio.get_event_loop()
bot = Bot(token=API_TOKEN, loop=loop)
dp = Dispatcher(bot)
@dp.message_handler(commands=['start'])
async def send_welcome(message: types.Message):
await message.reply("Hi!\nI'm EchoBot!\nPowered by aiogram.")
@dp.message_handler(commands=['sticker'])
async def save_sticker(message: types.Message):
async def handle_bad_message(msg: types.Message):
"""
Handler for unknown messages
"""
await msg.reply('That is not a sticker!')
port = args.port
cert = args.cert
pkey = args.pkey
host_name = args.host_name or host
webhook_port = args.webhook_port or port
webhook_path = args.webhook_path
# Fi webhook path
if not webhook_path.startswith("/"):
webhook_path = "/" + webhook_path
# Generate webhook URL
webhook_url = f"https://{host_name}:{webhook_port}{webhook_path}"
# Create bot & dispatcher instances.
bot = Bot(token)
dispatcher = Dispatcher(bot)
if (sock or host) and host_name:
if cert and pkey:
ssl_context = ssl.SSLContext(ssl.PROTOCOL_TLSv1_2)
ssl_context.load_cert_chain(cert, pkey)
else:
ssl_context = None
start_webhook(
dispatcher,
webhook_path,
on_startup=functools.partial(on_startup, url=webhook_url, cert=cert),
on_shutdown=on_shutdown,
host=host,
port=port,
from pathlib import Path
from aiogram import Bot, Dispatcher, types
from aiogram.contrib.fsm_storage.redis import RedisStorage2
from loguru import logger
from app import config
from app.middlewares.i18n import I18nMiddleware
app_dir: Path = Path(__file__).parent.parent
locales_dir = app_dir / "locales"
bot = Bot(config.TELEGRAM_TOKEN, parse_mode=types.ParseMode.HTML)
storage = RedisStorage2(host=config.REDIS_HOST, port=config.REDIS_PORT, db=config.REDIS_DB_FSM)
dp = Dispatcher(bot, storage=storage)
i18n = I18nMiddleware("bot", locales_dir, default="en")
def setup():
from app import filters
from app import middlewares
from app.utils import executor
middlewares.setup(dp)
filters.setup(dp)
executor.setup()
logger.info("Configure handlers...")
# noinspection PyUnresolvedReferences