# Pyrogram - Telegram MTProto API Client Library for Python

# Copyright (C) 2017-2019 Dan Tès <https://github.com/delivrance>

#

# This file is part of Pyrogram.

#

# Pyrogram is free software: you can redistribute it and/or modify

# it under the terms of the GNU Lesser General Public License as published

# by the Free Software Foundation, either version 3 of the License, or

# (at your option) any later version.

#

# Pyrogram is distributed in the hope that it will be useful,

# but WITHOUT ANY WARRANTY; without even the implied warranty of

# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the

# GNU Lesser General Public License for more details.

#

# You should have received a copy of the GNU Lesser General Public License

# along with Pyrogram. If not, see <http://www.gnu.org/licenses/>.

import logging

import math

import os

import re

import shutil

import tempfile

import threading

import time

from configparser import ConfigParser

from hashlib import sha256, md5

from importlib import import_module

from pathlib import Path

from signal import signal, SIGINT, SIGTERM, SIGABRT

from threading import Thread

from typing import Union, List

from pyrogram.api import functions, types

from pyrogram.api.core import TLObject

from pyrogram.client.handlers import DisconnectHandler

from pyrogram.client.handlers.handler import Handler

from pyrogram.client.methods.password.utils import compute_check

from pyrogram.crypto import AES

from pyrogram.errors import (

PhoneMigrate, NetworkMigrate, SessionPasswordNeeded,

FloodWait, PeerIdInvalid, VolumeLocNotFound, UserMigrate, ChannelPrivate, AuthBytesInvalid,

BadRequest)

from pyrogram.session import Auth, Session

from .ext import utils, Syncer, BaseClient, Dispatcher

from .methods import Methods

from .storage import Storage, FileStorage, MemoryStorage

from .types import User, SentCode, TermsOfService

log = logging.getLogger(__name__)

class Client(Methods, BaseClient):

"""Pyrogram Client, the main means for interacting with Telegram.

Parameters:

session_name (``str``):

Pass a string of your choice to give a name to the client session, e.g.: "*my_account*". This name will be

used to save a file on disk that stores details needed to reconnect without asking again for credentials.

Alternatively, if you don't want a file to be saved on disk, pass the special name "**:memory:**" to start

an in-memory session that will be discarded as soon as you stop the Client. In order to reconnect again

using a memory storage without having to login again, you can use

:meth:`~pyrogram.Client.export_session_string` before stopping the client to get a session string you can

pass here as argument.

api_id (``int`` | ``str``, *optional*):

The *api_id* part of your Telegram API Key, as integer. E.g.: "12345".

This is an alternative way to pass it if you don't want to use the *config.ini* file.

api_hash (``str``, *optional*):

The *api_hash* part of your Telegram API Key, as string. E.g.: "0123456789abcdef0123456789abcdef".

This is an alternative way to set it if you don't want to use the *config.ini* file.

app_version (``str``, *optional*):

Application version. Defaults to "Pyrogram |version|".

This is an alternative way to set it if you don't want to use the *config.ini* file.

device_model (``str``, *optional*):

Device model. Defaults to *platform.python_implementation() + " " + platform.python_version()*.

This is an alternative way to set it if you don't want to use the *config.ini* file.

system_version (``str``, *optional*):

Operating System version. Defaults to *platform.system() + " " + platform.release()*.

This is an alternative way to set it if you don't want to use the *config.ini* file.

lang_code (``str``, *optional*):

Code of the language used on the client, in ISO 639-1 standard. Defaults to "en".

This is an alternative way to set it if you don't want to use the *config.ini* file.

ipv6 (``bool``, *optional*):

Pass True to connect to Telegram using IPv6.

Defaults to False (IPv4).

proxy (``dict``, *optional*):

Your SOCKS5 Proxy settings as dict,

e.g.: *dict(hostname="11.22.33.44", port=1080, username="user", password="pass")*.

The *username* and *password* can be omitted if your proxy doesn't require authorization.

This is an alternative way to setup a proxy if you don't want to use the *config.ini* file.

test_mode (``bool``, *optional*):

Enable or disable login to the test servers.

Only applicable for new sessions and will be ignored in case previously created sessions are loaded.

Defaults to False.

bot_token (``str``, *optional*):

Pass your Bot API token to create a bot session, e.g.: "123456:ABC-DEF1234ghIkl-zyx57W2v1u123ew11"

Only applicable for new sessions.

This is an alternative way to set it if you don't want to use the *config.ini* file.

phone_number (``str``, *optional*):

Pass your phone number as string (with your Country Code prefix included) to avoid entering it manually.

Only applicable for new sessions.

phone_code (``str``, *optional*):

Pass the phone code as string (for test numbers only) to avoid entering it manually.

Only applicable for new sessions.

password (``str``, *optional*):

Pass your Two-Step Verification password as string (if you have one) to avoid entering it manually.

Only applicable for new sessions.

force_sms (``bool``, *optional*):

Pass True to force Telegram sending the authorization code via SMS.

Only applicable for new sessions.

Defaults to False.

workers (``int``, *optional*):

Thread pool size for handling incoming updates.

Defaults to 4.

workdir (``str``, *optional*):

Define a custom working directory. The working directory is the location in your filesystem where Pyrogram

will store your session files.

Defaults to the parent directory of the main script.

config_file (``str``, *optional*):

Path of the configuration file.

Defaults to ./config.ini

plugins (``dict``, *optional*):

Your Smart Plugins settings as dict, e.g.: *dict(root="plugins")*.

This is an alternative way setup plugins if you don't want to use the *config.ini* file.

no_updates (``bool``, *optional*):

Pass True to completely disable incoming updates for the current session.

When updates are disabled your client can't receive any new message.

Useful for batch programs that don't need to deal with updates.

Defaults to False (updates enabled and always received).

takeout (``bool``, *optional*):

Pass True to let the client use a takeout session instead of a normal one, implies *no_updates=True*.

Useful for exporting your Telegram data. Methods invoked inside a takeout session (such as get_history,

download_media, ...) are less prone to throw FloodWait exceptions.

Only available for users, bots will ignore this parameter.

Defaults to False (normal session).

"""

def __init__(

self,

session_name: Union[str, Storage],

api_id: Union[int, str] = None,

api_hash: str = None,

app_version: str = None,

device_model: str = None,

system_version: str = None,

lang_code: str = None,

ipv6: bool = False,

proxy: dict = None,

test_mode: bool = False,

bot_token: str = None,

phone_number: str = None,

phone_code: str = None,

password: str = None,

force_sms: bool = False,

workers: int = BaseClient.WORKERS,

workdir: str = BaseClient.WORKDIR,

config_file: str = BaseClient.CONFIG_FILE,

plugins: dict = None,

no_updates: bool = None,

takeout: bool = None

):

super().__init__()

self.session_name = session_name

self.api_id = int(api_id) if api_id else None

self.api_hash = api_hash

self.app_version = app_version

self.device_model = device_model

self.system_version = system_version

self.lang_code = lang_code

self.ipv6 = ipv6

# TODO: Make code consistent, use underscore for private/protected fields

self._proxy = proxy

self.test_mode = test_mode

self.bot_token = bot_token

self.phone_number = phone_number

self.phone_code = phone_code

self.password = password

self.force_sms = force_sms

self.workers = workers

self.workdir = Path(workdir)

self.config_file = Path(config_file)

self.plugins = plugins

self.no_updates = no_updates

self.takeout = takeout

if isinstance(session_name, str):

if session_name == ":memory:" or len(session_name) >= MemoryStorage.SESSION_STRING_SIZE:

session_name = re.sub(r"[\n\s]+", "", session_name)

self.storage = MemoryStorage(session_name)

else:

self.storage = FileStorage(session_name, self.workdir)

elif isinstance(session_name, Storage):

self.storage = session_name

else:

raise ValueError("Unknown storage engine")

self.dispatcher = Dispatcher(self, workers)

def __enter__(self):

return self.start()

def __exit__(self, *args):

try:

self.stop()

except ConnectionError:

pass

@property

def proxy(self):

return self._proxy

@proxy.setter

def proxy(self, value):

if value is None:

self._proxy = None

return

if self._proxy is None:

self._proxy = {}

self._proxy["enabled"] = bool(value.get("enabled", True))

self._proxy.update(value)

def connect(self) -> bool:

"""

Connect the client to Telegram servers.

Returns:

``bool``: On success, in case the passed-in session is authorized, True is returned. Otherwise, in case

the session needs to be authorized, False is returned.

Raises:

ConnectionError: In case you try to connect an already connected client.

"""

if self.is_connected:

raise ConnectionError("Client is already connected")

self.load_config()

self.load_session()

self.session = Session(self, self.storage.dc_id(), self.storage.auth_key())

self.session.start()

self.is_connected = True

return bool(self.storage.user_id())

def disconnect(self):

"""Disconnect the client from Telegram servers.

Raises:

ConnectionError: In case you try to disconnect an already disconnected client or in case you try to

disconnect a client that needs to be terminated first.

"""

if not self.is_connected:

raise ConnectionError("Client is already disconnected")

if self.is_initialized:

raise ConnectionError("Can't disconnect an initialized client")

self.session.stop()

self.storage.close()

self.is_connected = False

def initialize(self):

"""Initialize the client by starting up workers.

This method will start updates and download workers.

It will also load plugins and start the internal dispatcher.

Raises:

ConnectionError: In case you try to initialize a disconnected client or in case you try to initialize an

already initialized client.

"""

if not self.is_connected:

raise ConnectionError("Can't initialize a disconnected client")

if self.is_initialized:

raise ConnectionError("Client is already initialized")

self.load_plugins()

for i in range(self.UPDATES_WORKERS):

self.updates_workers_list.append(

Thread(

target=self.updates_worker,

name="UpdatesWorker#{}".format(i + 1)

)

)

self.updates_workers_list[-1].start()

for i in range(self.DOWNLOAD_WORKERS):

self.download_workers_list.append(

Thread(

target=self.download_worker,

name="DownloadWorker#{}".format(i + 1)

)

)

self.download_workers_list[-1].start()

self.dispatcher.start()

Syncer.add(self)

self.is_initialized = True

def terminate(self):

"""Terminate the client by shutting down workers.

This method does the opposite of :meth:`~Client.initialize`.

It will stop the dispatcher and shut down updates and download workers.

Raises:

ConnectionError: In case you try to terminate a client that is already terminated.

"""

if not self.is_initialized:

raise ConnectionError("Client is already terminated")

if self.takeout_id:

self.send(functions.account.FinishTakeoutSession())

log.warning("Takeout session {} finished".format(self.takeout_id))

Syncer.remove(self)

self.dispatcher.stop()

for _ in range(self.DOWNLOAD_WORKERS):

self.download_queue.put(None)

for i in self.download_workers_list:

i.join()

self.download_workers_list.clear()

for _ in range(self.UPDATES_WORKERS):

self.updates_queue.put(None)

for i in self.updates_workers_list:

i.join()

self.updates_workers_list.clear()

for i in self.media_sessions.values():

i.stop()

self.media_sessions.clear()

self.is_initialized = False

def send_code(self, phone_number: str) -> SentCode:

"""Send the confirmation code to the given phone number.

Parameters:

phone_number (``str``):

Phone number in international format (includes the country prefix).

Returns:

:obj:`SentCode`: On success, an object containing information on the sent confirmation code is returned.

Raises:

BadRequest: In case the phone number is invalid.

"""

phone_number = phone_number.strip(" +")

while True:

try:

r = self.send(

functions.auth.SendCode(

phone_number=phone_number,

api_id=self.api_id,

api_hash=self.api_hash,

settings=types.CodeSettings()

)

)

except (PhoneMigrate, NetworkMigrate) as e:

self.session.stop()

self.storage.dc_id(e.x)

self.storage.auth_key(Auth(self, self.storage.dc_id()).create())

self.session = Session(self, self.storage.dc_id(), self.storage.auth_key())

self.session.start()

else:

return SentCode._parse(r)

def resend_code(self, phone_number: str, phone_code_hash: str) -> SentCode:

"""Re-send the confirmation code using a different type.

The type of the code to be re-sent is specified in the *next_type* attribute of the :obj:`SentCode` object

returned by :meth:`send_code`.

Parameters:

phone_number (``str``):

Phone number in international format (includes the country prefix).

phone_code_hash (``str``):

Confirmation code identifier.

Returns:

:obj:`SentCode`: On success, an object containing information on the re-sent confirmation code is returned.

Raises:

BadRequest: In case the arguments are invalid.

"""

phone_number = phone_number.strip(" +")

r = self.send(

functions.auth.ResendCode(

phone_number=phone_number,

phone_code_hash=phone_code_hash

)

)

return SentCode._parse(r)

def sign_in(self, phone_number: str, phone_code_hash: str, phone_code: str) -> Union[User, TermsOfService, bool]:

"""Authorize a user in Telegram with a valid confirmation code.

Parameters:

phone_number (``str``):

Phone number in international format (includes the country prefix).

phone_code_hash (``str``):

Code identifier taken from the result of :meth:`~Client.send_code`.

phone_code (``str``):

The valid confirmation code you received (either as Telegram message or as SMS in your phone number).

Returns:

:obj:`User` | :obj:`TermsOfService` | bool: On success, in case the authorization completed, the user is

returned. In case the phone number needs to be registered first AND the terms of services accepted (with

:meth:`~Client.accept_terms_of_service`), an object containing them is returned. In case the phone number

needs to be registered, but the terms of services don't need to be accepted, False is returned instead.

Raises:

BadRequest: In case the arguments are invalid.

SessionPasswordNeeded: In case a password is needed to sign in.

"""

phone_number = phone_number.strip(" +")

r = self.send(

functions.auth.SignIn(

phone_number=phone_number,

phone_code_hash=phone_code_hash,

phone_code=phone_code

)

)

if isinstance(r, types.auth.AuthorizationSignUpRequired):

if r.terms_of_service:

return TermsOfService._parse(terms_of_service=r.terms_of_service)

return False

else:

self.storage.user_id(r.user.id)

self.storage.is_bot(False)

return User._parse(self, r.user)

def sign_up(self, phone_number: str, phone_code_hash: str, first_name: str, last_name: str = "") -> User:

"""Register a new user in Telegram.

Parameters:

phone_number (``str``):

Phone number in international format (includes the country prefix).

phone_code_hash (``str``):

Code identifier taken from the result of :meth:`~Client.send_code`.

first_name (``str``):

New user first name.

last_name (``str``, *optional*):

New user last name. Defaults to "" (empty string).

Returns:

:obj:`User`: On success, the new registered user is returned.

Raises:

BadRequest: In case the arguments are invalid.

"""

phone_number = phone_number.strip(" +")

r = self.send(

functions.auth.SignUp(

phone_number=phone_number,

first_name=first_name,

last_name=last_name,

phone_code_hash=phone_code_hash

)

)

self.storage.user_id(r.user.id)

self.storage.is_bot(False)

return User._parse(self, r.user)

def sign_in_bot(self, bot_token: str) -> User:

"""Authorize a bot using its bot token generated by BotFather.

Parameters:

bot_token (``str``):

The bot token generated by BotFather

Returns:

:obj:`User`: On success, the bot identity is return in form of a user object.

Raises:

BadRequest: In case the bot token is invalid.

"""

while True:

try:

r = self.send(

functions.auth.ImportBotAuthorization(

flags=0,

api_id=self.api_id,

api_hash=self.api_hash,

bot_auth_token=bot_token

)

)

except UserMigrate as e:

self.session.stop()

self.storage.dc_id(e.x)

self.storage.auth_key(Auth(self, self.storage.dc_id()).create())

self.session = Session(self, self.storage.dc_id(), self.storage.auth_key())

self.session.start()

else:

self.storage.user_id(r.user.id)

self.storage.is_bot(True)

return User._parse(self, r.user)

def get_password_hint(self) -> str:

"""Get your Two-Step Verification password hint.

Returns:

``str``: On success, the password hint as string is returned.

"""

return self.send(functions.account.GetPassword()).hint

def check_password(self, password: str) -> User:

"""Check your Two-Step Verification password and log in.

Parameters:

password (``str``):

Your Two-Step Verification password.

Returns:

:obj:`User`: On success, the authorized user is returned.

Raises:

BadRequest: In case the password is invalid.

"""

r = self.send(

functions.auth.CheckPassword(

password=compute_check(

self.send(functions.account.GetPassword()),

password

)

)

)

self.storage.user_id(r.user.id)

self.storage.is_bot(False)

return User._parse(self, r.user)

def send_recovery_code(self) -> str:

"""Send a code to your email to recover your password.

Returns:

``str``: On success, the hidden email pattern is returned and a recovery code is sent to that email.

Raises:

BadRequest: In case no recovery email was set up.

"""

return self.send(

functions.auth.RequestPasswordRecovery()

).email_pattern

def recover_password(self, recovery_code: str) -> User:

"""Recover your password with a recovery code and log in.

Parameters:

recovery_code (``str``):

The recovery code sent via email.

Returns:

:obj:`User`: On success, the authorized user is returned and the Two-Step Verification password reset.

Raises:

BadRequest: In case the recovery code is invalid.

"""

r = self.send(

functions.auth.RecoverPassword(

code=recovery_code

)

)

self.storage.user_id(r.user.id)

self.storage.is_bot(False)

return User._parse(self, r.user)

def accept_terms_of_service(self, terms_of_service_id: str) -> bool:

"""Accept the given terms of service.

Parameters:

terms_of_service_id (``str``):

The terms of service identifier.

"""

r = self.send(

functions.help.AcceptTermsOfService(

id=types.DataJSON(

data=terms_of_service_id

)

)

)

assert r

return True

def authorize(self) -> User:

if self.bot_token:

return self.sign_in_bot(self.bot_token)

while True:

try:

if not self.phone_number:

while True:

value = input("Enter phone number or bot token: ")

if not value:

continue

confirm = input("Is \"{}\" correct? (y/N): ".format(value)).lower()

if confirm == "y":

break

if ":" in value:

self.bot_token = value

return self.sign_in_bot(value)

else:

self.phone_number = value

sent_code = self.send_code(self.phone_number)

except BadRequest as e:

print(e.MESSAGE)

self.phone_number = None

self.bot_token = None

except FloodWait as e:

print(e.MESSAGE.format(x=e.x))

time.sleep(e.x)

else:

break

if self.force_sms:

sent_code = self.resend_code(self.phone_number, sent_code.phone_code_hash)

print("The confirmation code has been sent via {}".format(

{

"app": "Telegram app",

"sms": "SMS",

"call": "phone call",

"flash_call": "phone flash call"

}[sent_code.type]

))

while True:

if not self.phone_code:

self.phone_code = input("Enter confirmation code: ")

try:

signed_in = self.sign_in(self.phone_number, sent_code.phone_code_hash, self.phone_code)

except BadRequest as e:

print(e.MESSAGE)

self.phone_code = None

except SessionPasswordNeeded as e:

print(e.MESSAGE)

while True:

print("Password hint: {}".format(self.get_password_hint()))

if not self.password:

self.password = input("Enter password (empty to recover): ")

try:

if not self.password:

confirm = input("Confirm password recovery (y/n): ")

if confirm == "y":

email_pattern = self.send_recovery_code()

print("The recovery code has been sent to {}".format(email_pattern))

while True:

recovery_code = input("Enter recovery code: ")

try:

return self.recover_password(recovery_code)

except BadRequest as e:

print(e.MESSAGE)

except FloodWait as e:

print(e.MESSAGE.format(x=e.x))

time.sleep(e.x)

except Exception as e:

log.error(e, exc_info=True)

raise

else:

self.password = None

else:

return self.check_password(self.password)

except BadRequest as e:

print(e.MESSAGE)

self.password = None

except FloodWait as e:

print(e.MESSAGE.format(x=e.x))

time.sleep(e.x)

except FloodWait as e:

print(e.MESSAGE.format(x=e.x))

time.sleep(e.x)

else:

break

if isinstance(signed_in, User):

return signed_in

while True:

first_name = input("Enter first name: ")

last_name = input("Enter last name (empty to skip): ")

try:

signed_up = self.sign_up(

self.phone_number,

sent_code.phone_code_hash,

first_name,

last_name

)

except BadRequest as e:

print(e.MESSAGE)

except FloodWait as e:

print(e.MESSAGE.format(x=e.x))

time.sleep(e.x)

else:

break

if isinstance(signed_in, TermsOfService):

print("\n" + signed_in.text + "\n")

self.accept_terms_of_service(signed_in.id)

return signed_up

def log_out(self):

"""Log out from Telegram and delete the *\\*.session* file.

When you log out, the current client is stopped and the storage session deleted.

No more API calls can be made until you start the client and re-authorize again.

Returns:

``bool``: On success, True is returned.

Example:

.. code-block:: python

# Log out.

app.log_out()

"""

self.send(functions.auth.LogOut())

self.stop()

self.storage.delete()

return True

def start(self):

"""Start the client.

This method connects the client to Telegram and, in case of new sessions, automatically manages the full

authorization process using an interactive prompt.

Returns:

:obj:`Client`: The started client itself.

Raises:

ConnectionError: In case you try to start an already started client.

Example:

.. code-block:: python

:emphasize-lines: 4

from pyrogram import Client

app = Client("my_account")

app.start()

... # Call API methods

app.stop()

"""

is_authorized = self.connect()

try:

if not is_authorized:

self.authorize()

if not self.storage.is_bot() and self.takeout:

self.takeout_id = self.send(functions.account.InitTakeoutSession()).id

log.warning("Takeout session {} initiated".format(self.takeout_id))

self.send(functions.updates.GetState())

except (Exception, KeyboardInterrupt):

self.disconnect()

raise

else:

self.initialize()

return self

def stop(self):

"""Stop the Client.

This method disconnects the client from Telegram and stops the underlying tasks.

Returns:

:obj:`Client`: The stopped client itself.

Raises:

ConnectionError: In case you try to stop an already stopped client.

Example:

.. code-block:: python

:emphasize-lines: 8

from pyrogram import Client

app = Client("my_account")

app.start()

... # Call API methods

app.stop()

"""

self.terminate()

self.disconnect()

return self

def restart(self):

"""Restart the Client.

This method will first call :meth:`~Client.stop` and then :meth:`~Client.start` in a row in order to restart

a client using a single method.

Returns:

:obj:`Client`: The restarted client itself.

Raises:

ConnectionError: In case you try to restart a stopped Client.

Example:

.. code-block:: python

:emphasize-lines: 8

from pyrogram import Client

app = Client("my_account")

app.start()

... # Call API methods

app.restart()

... # Call other API methods

app.stop()

"""

self.stop()

self.start()

return self

@staticmethod

def idle(stop_signals: tuple = (SIGINT, SIGTERM, SIGABRT)):

"""Block the main script execution until a signal is received.

This static method will run an infinite loop in order to block the main script execution and prevent it from

exiting while having client(s) that are still running in the background.

It is useful for event-driven application only, that are, applications which react upon incoming Telegram

updates through handlers, rather than executing a set of methods sequentially.

The way Pyrogram works, it will keep your handlers in a pool of worker threads, which are executed concurrently

outside the main thread; calling idle() will ensure the client(s) will be kept alive by not letting the main

script to end, until you decide to quit.

Once a signal is received (e.g.: from CTRL+C) the inner infinite loop will break and your main script will

continue. Don't forget to call :meth:`~Client.stop` for each running client before the script ends.

Parameters:

stop_signals (``tuple``, *optional*):

Iterable containing signals the signal handler will listen to.

Defaults to *(SIGINT, SIGTERM, SIGABRT)*.

Example:

.. code-block:: python

:emphasize-lines: 13

from pyrogram import Client

app1 = Client("account1")

app2 = Client("account2")

app3 = Client("account3")

... # Set handlers up

app1.start()

app2.start()

app3.start()

Client.idle()

app1.stop()

app2.stop()

app3.stop()

"""

def signal_handler(_, __):

Client.is_idling = False

for s in stop_signals:

signal(s, signal_handler)

Client.is_idling = True

while Client.is_idling:

time.sleep(1)

def run(self):

"""Start the client, idle the main script and finally stop the client.

This is a convenience method that calls :meth:`~Client.start`, :meth:`~Client.idle` and :meth:`~Client.stop` in

sequence. It makes running a client less verbose, but is not suitable in case you want to run more than one

client in a single main script, since idle() will block after starting the own client.

Raises:

ConnectionError: In case you try to run an already started client.

Example:

.. code-block:: python

:emphasize-lines: 7

from pyrogram import Client

app = Client("my_account")

... # Set handlers up

app.run()

"""

self.start()

Client.idle()

self.stop()

def add_handler(self, handler: Handler, group: int = 0):

"""Register an update handler.

You can register multiple handlers, but at most one handler within a group will be used for a single update.

To handle the same update more than once, register your handler using a different group id (lower group id

== higher priority). This mechanism is explained in greater details at

:doc:`More on Updates <../../topics/more-on-updates>`.

Parameters:

handler (``Handler``):

The handler to be registered.

group (``int``, *optional*):