tgbox package

tgbox package#

tgbox.api.local module#

Module with API functions and classes for LocalBox.

class tgbox.api.local.DecryptedLocalBox(elb: EncryptedLocalBox, key: BaseKey | MainKey)[source]#

Bases: EncryptedLocalBox

Class that represents decrypted local box. On more low-level it’s wrapper around TgboxDB that decrypts and parses every row. You don’t need to work with EncryptedLocalBox to write any data to the TgboxDB. Every commit will be encrypted here.

Typical usage:

property api_hash: str | None#: Returns API_HASH.

property api_id: int | None#: Returns API_ID.

property box_channel_id: bytes | int | None#: Returns encrypted channel ID from EncryptedLocalBox and decrypted from DecryptedLocalBox.

property box_cr_time: bytes | int | None#: Returns encrypted box creation time from EncryptedLocalBox and decrypted from DecryptedLocalBox.

property box_salt: BoxSalt | None#: Returns BoxSalt or None if not initialized

async contents(sfpid: bytes | None = None, ignore_files: bool | None = False, erase_encrypted_metadata: bool | None = True) → AsyncGenerator[EncryptedLocalBoxDirectory | DecryptedLocalBoxDirectory, None]#

Recursive iterate over all files/folders in LocalBox.

Arguments:

sfpid (bytes, optional):: Will start from this PartID if specified, else will start from “root” PIDs.
ignore_files (bool, optional):: Will not return LocalBoxFile associated with the LocalBoxDirectory if False.
erase_encrypted_metadata (bool, optional):: Will remove metadata from the parent EncryptedLocalBoxFile after decryption to save more RAM if True. You can call .init() method on it to load it again.

async static decrypt() → NoReturn[source]#

Will return DecryptedLocalBox.

You should specify BaseKey if you want to access session and use this dlb with RemoteBox, however, you can specify MainKey if you only want to iterate over local files / fetch basic local info.

property defaults: DefaultsTableWrapper | RemoteBoxDefaults#: Returns DefaultsTableWrapper or RemoteBoxDefaults.

async delete() → None#

This method WILL DELETE your LocalBox database. It doesn’t affect RemoteBox, so you can make new LocalBox from the Remote version if you have MainKey.

Will raise FileNotFoundError if something goes wrong (i.e DB was moved).

async delete_files(*lbf: 'EncryptedLocalBoxFile' | 'DecryptedLocalBoxFile', lbf_ids: list | None = None, rb: 'tgbox.api.remote.EncryptedRemoteBox' | 'tgbox.api.remote.DecryptedRemoteBox' | None = None, remove_empty_directories: bool | None = False) → None#

A function to remove a bunch of local files at once.

Arguments:

lbf (EncryptedLocalBoxFile, DecryptedLocalBoxFile, asterisk):

(Encrypted|Decrypted)LocalBoxFile(s) to remove.

lbf_ids (list, optional):

You can specify ids instead of LocalBox file objects. However, lbf is preferred here.

rb (EncryptedRemoteBox, DecryptedRemoteBox, optional):

You can specify a RemoteBox associated with current LocalBox to also remove all specified files in RemoteBox too.

remove_empty_directories (bool, optional):

If True, will remove orphaned directories that left after removing files (if any).

Alternative: dlb.remove_empty_directories()

Note

Without rb this will delete files only from your LocalBox. To completely remove your file use the same coroutine on RemoteBox or specify (Encrypted|Decrypted)RemoteBox.

async done()#: Await this method when you end all work with LocalBox, so we will clean up & close connections.

Yields every local file as EncryptedLocalBoxFile if you call it on EncryptedLocalBox and DecryptedLocalBoxFile if on DecryptedLocalBox. Works via self.get_file

Alternatively, you may use LocalBox.contents method.

Arguments:

cache_preview (bool, optional):: Cache preview in class or not.
min_id (int, optional):: Will iterate from this ID.
max_id (int, optional):: Will iterate up to this ID.
ids (int, list, optional):: ID or list with IDs you want to fetch. If specified, The min_id and max_id args will be ignored
decrypt (bool, optional):: Will return EncryptedLocalBoxFile if False, and DecryptedLocalBoxFile if True. If None, will be determined by class.
reverse (bool, optional):: If set to True, the local files will be returned in reverse order (from newest to oldest, instead of the default oldest to newest).
fetch_count (int, optional):: Amount of files generator will fetch and cache from SQLite table before return. 100 by default.
erase_encrypted_metadata (bool, optional):: Will remove metadata from the parent EncryptedLocalBoxFile after decryption to save more RAM if True. You can call .init() method on it to load it again.

async get_directory(path: Path | str) → DecryptedLocalBoxDirectory | None[source]#

This method will make DecryptedLocalBoxDirectory from your path. If such path is not presented in LocalBox, then None will be returned.

Arguments:

path (Path, str):: Absolute path from which you want to make an DecryptedLocalBoxDirectory.

Returns EncryptedLocalBoxFile from EncryptedLocalBox or DecryptedLocalBoxFile from DecryptedLocalBox if file exists. None otherwise.

Arguments:

id (int, optional):: File ID. Must be specified if fingerprint is None.
fingerprint (bytes, optional):: File Fingerprint. Must be specified if id argument is None.
decrypt (bool, optional):: Will return EncryptedLocalBoxFile if False, and DecryptedLocalBoxFile if True. If None, will be determined by class.
cache_preview (bool, optional):: Cache preview in class or not.
erase_encrypted_metadata (bool, optional):: Will remove metadata from the parent EncryptedLocalBoxFile after decryption to save more RAM if True. You can call .init() method on it to load it again.

async get_files_total() → int#: Returns a total number of files in this LocalBox

async get_last_file_id() → int#: Returns last file id from the FILES table. If there is no files at all, will return 0.

get_requestkey(basekey: BaseKey) → RequestKey#

Returns RequestKey for this LocalBox. You should use this method if you want to decrypt other’s RemoteBox.

Arguments:

basekey (BaseKey):: To make a RequestKey for other’s RemoteBox you need to create new BaseKey for it. Later this key will be used for Box decryption.

get_sharekey(reqkey: RequestKey | None = None) → ShareKey[source]#

Returns ShareKey for this Box. You should use this method if you want to share your LocalBox with other people.

Arguments:

reqkey (RequestKey, optional):: Requester’s RequestKey. If isn’t specified returns ImportKey of this box without encryption, so anyone with this key can decrypt ALL files in your Boxes.

async import_file(drbf: tgbox.api.remote.DecryptedRemoteBoxFile, file_path: str | Path | None = None) → DecryptedLocalBoxFile[source]#

Imports file to your DecryptedLocalBox

Arguments:

drbf (DecryptedRemoteBoxFile):

Remote file you want to import.

file_path (Path, optional):

File’s path. Will be used drbf._file_path if None and if drbf was decrypted with the MainKey, otherwise self.defaults.DEF_NO_FOLDER.

This method will call a set_file_path(file_path) on the specified drbf if file_path isn’t a None.

You can change drbf file_path with set_file_path method before importing file, so you don’t need to specify it here.

async static init() → NoReturn[source]#: Will fetch and parse data from Database.

property initialized: bool#: Returns True if you called .init()

property is_encrypted: bool#: Will return True if this is an Encrypted class, False if Decrypted

property mainkey: MainKey#: Will return MainKey of this Box

Prepares your file for RemoteBox.push_file

Arguments:

file (BinaryIO, bytes, TelegramVirtualFile):

file data to add to the LocalBox. In most cases it’s just opened file. If you want to upload something else, then you need to implement class that have read, name and basic seek() methods. (support for seek(0,0) is enough).

The method needs to know size of the file, so it will try to ask system what size of file on path file.name. If it’s impossible, method will try to seek file to EOF, if file isn’t seekable, we will try to get size by len() (as __len__ dunder). If all this fails, method will try to get length of file.read()) (with load to RAM!!!).

Absolute file path length must be <= self.defaults.FILE_PATH_MAX; If file has no name and file_path is not specified, – it will be NO_FOLDER/{tgbox.tools.prbg(6).hex()}.

file_size (int, optional):

Bytelength of file. You can specify it if you already know file size.

file_path (str, Path, optional):

File path of Box file (file name must be included). If not specified, will be used path from the BinaryIO, (file arg) if file is not a BinaryIO then will be used a self.defaults.DEF_NO_FOLDER.

Must be <= self.defaults.FILE_PATH_MAX.

cattrs (Dict[str, Union[bytes, None]], optional):

The file’s custom metadata attributes that will be added to the main metadata. Specified dict will be packed with the PackedAttributes.

Please note that after some operations we will create a file metadata. Its limit defined as self.defaults.METADATA_MAX. You shouldn’t overflow this number plus size of default metadata; other is up to you.

make_preview (bool, optional):

Will try to add file preview to the metadata if True (default).

skip_fingerprint_check (bool, optional):

If True, will skip the File Fingerprint check. Change it only if you want to update already pushed to RemoteBox file.

async remove_empty_directories(part_ids: List[bytes] | None = None)#

By default, the Protocol will not automatically remove empty Directories after the file deletion process was done (unless forced). To do this, use this method.

Arguments:

part_ids (List[bytes], optional):

List of Part ID you want to check. Will remove only if Directory attached to this Part ID is orphaned (no files/other Directories linked)

If not specified, will check every Part ID in LocalBox.

async replace_session(basekey: BaseKey, tc: TelegramClient) → None[source]#

This method will replace LocalBox session to session of specified TelegramClient.

Arguments:

basekey (BaseKey):: BaseKey of this LocalBox.
tc (TelegramClient):: TelegramClient from which we will extract new session.

async search_file(sf: SearchFilter, cache_preview: bool = True, reverse: bool = False, fetch_count: int = 100, erase_encrypted_metadata: bool = True) → AsyncGenerator[DecryptedLocalBoxFile, None][source]#

Use this method search for files in your DecryptedLocalBox.

Arguments:

sf (SearchFilter):: SearchFilter with kwargs you like.
cache_preview (bool, optional):: Will cache preview in file object if True.
reverse (bool, optional):: If set to True, the local files will be searched in reverse order (from newest to oldest, instead of the default oldest to newest).
fetch_count (int, optional):: Amount of files generator will fetch and cache from SQLite table before return. 100 by default.
erase_encrypted_metadata (bool, optional):: Will remove metadata from the parent EncryptedLocalBoxFile after decryption to save more RAM if True. You can call .init() method on it to load it again.

property session: bytes | str | None#: Returns encrypted session from EncryptedLocalBox and decrypted from DecryptedLocalBox.

async sync(drb: tgbox.api.remote.DecryptedRemoteBox, deep: bool | None = False, start_from: int | None = 0, timeout: int | None = 15, deep_progress_callback: Callable[[int, int], None] | None = None, fast_progress_callback: Callable[[int, str], None] | None = None)[source]#

This method will synchronize your LocalBox with RemoteBox. All files that not in RemoteBox but in Local will be removed, all that in Remote but not in LocalBox will be imported.

drb (DecryptedRemoteBox):

RemoteBox associated with this LocalBox.

deep (bool, optional):

Flag to enable a “deep syncing”.

start_from (int, optional):

Will check files that > start_from [ID]. Will be used only on deep syncing.

timeout (int, optional):

How many seconds generator will sleep at every 1000 file. By default it’s 15 seconds. Don’t use too low timeouts or you will receive FloodWaitError. Will be used only on Deep Sync, Fast Sync will ignore this argument.

deep_progress_callback (Callable[[int, int], None], optional):

A callback function accepting two parameters: (current_id, last_id). Will be used only on deep syncing.

fast_progress_callback (Callable[[int, str], None], optional):

A callback function accepting two parameters: (file_id, action<str>). Will be used only on fast syncing.

Don’t treat this as progressbar. The code will call/await the specified callback with one of the arguments from below:

fast_progress_callback(22, 'deleted') OR
fast_progress_callback(22, 'imported') OR
fast_progress_callback(22, 'metadata updated')

Note

By default this method will use a fast syncing, from the “Recent Actions” admin log. This is the best for changes made within 48 hours & useless after. Deep syncing will iterate over each file in the remote and compare it to local, thus, may take a very long time to complete.
In fast syncing we will fetch updates to Box only from other admins.

property tgbox_db: TgboxDB#: Returns TgboxDB.

class tgbox.api.local.DecryptedLocalBoxDirectory(elbd: EncryptedLocalBoxDirectory, key: BaseKey | MainKey | None = None, dlb: DecryptedLocalBox | None = None)[source]#

Bases: EncryptedLocalBoxDirectory

async static decrypt() → NoReturn[source]#

Decrypt self and return DecryptedLocalBoxDirectory

Arguments:

key (BaseKey, MainKey, optional):: Decryption key. Must be specified if dlb is None.
dlb (DecryptedLocalBox, optional):: DecryptedLocalBox that we will use to decrypt EncryptedLocalBoxDirectory. Must be specified if key argument is None.

async delete() → None#: Will delete this directory with all sub-dirs and files from your LocalBox. All of them will stay in RemoteBox, so you can restore all your data by syncing Box.

property floaded: bool#: Will return True if path was fully loaded

async get_contents_total() → int#: Will return a total number of contents in this directory

async get_files_total() → int#: Will return a total number of files in this directory

async get_folders_total() → int#: Will return a total number of folders in this directory

get_sharekey(reqkey: RequestKey | None = None) → ShareKey[source]#

Returns ShareKey for this file. You should use this method if you want to share ALL files from this directory with other user.

Use the same method on the DecryptedLocalBoxFile to share only one file with some Requester.

Arguments:

reqkey (RequestKey, optional):: Requester’s RequestKey. If isn’t specified, returns ShareKey of this directory without encryption, so ANYONE with this key can decrypt files from this Directory in Local & Remote.

async static init() → NoReturn[source]#: Will fetch required data from the database.

property initialized: bool#: Returns True if you already called .init()

property is_encrypted: bool#: Will return True if this is an Encrypted class, False if Decrypted

async iterdir(ignore_dirs: bool = False, ignore_files: bool = False, cache_preview: bool = True, ppid: bytes | DirectoryRoot | None = None, erase_encrypted_metadata: bool | None = True) → EncryptedLocalBoxFile | DecryptedLocalBoxFile | EncryptedLocalBoxDirectory | DecryptedLocalBoxDirectory#

Iterate over all files/folders in this abstract directory.

Arguments:

ignore_dirs (bool, optional):: Return abstract folders from this directory as LocalBoxDirectory or not. If False will return only LocalBoxFile objects.
ignore_files (bool, optional):: Will return LocalBoxFile if False.
cache_preview (bool, optional):: Cache preview in class or not. True by default.
ppid (bytes, DirectoryRoot):: Path PartID to iterate in. Will iterate over absolute LocalBox directory root if it’s DirectoryRoot. Will use self.part_id if not specified (by default).
erase_encrypted_metadata (bool, optional):: Will remove metadata from the parent EncryptedLocalBoxFile after decryption to save more RAM if True. You can call .init() method on it to load it again.

async lload(full: bool | None = False)#

Will load and return one path part from the left side (previous) or return None if there is nothing left.

Arguments:

full (bool, optional):: If full is True, will load all parts from the left side and return None.

property parent_part_id: bytes | None#: Returns None if you didn’t called the .init() method, parent path part ID otherwise.

property part: bytes | None#: Returns None if you didn’t called the .init() method, encrypted path part if you’re in EncryptedLocalBoxDirectory, and decrypted if DecryptedLocalBoxDirectory.

property part_id: bytes#: Returns path part ID

property parts: list#: Will return loaded path parts.

property tgbox_db: TgboxDB#: Will return TgboxDB associated with this directory

Bases: EncryptedLocalBoxFile

This class represents an decrypted local file. On more low-level that’s a wrapper of FILES table in Tgbox Database that decrypts row.

Typical usage:

from asyncio import run as asyncio_run
from tgbox.api import get_localbox
from tgbox.keys import make_basekey, Phrase

phrase = Phrase(b'example phrase here')
basekey = make_basekey(phrase)

async def main():
    dlb = await get_localbox(basekey)
    lfid = dlb.get_last_file_id()
    dlbfi = await dlb.get_file(lfid)

    print(dlbfi.id, dlbfi.id, dlbfi.size)

asyncio_run(main())

property box_salt: BoxSalt | None#: Returns box salt or None if file wasn’t initialized

property cattrs: bytes | None#: Returns file Custom Attributes

async static decrypt() → NoReturn[source]#

Returns decrypted by key/dlb EncryptedLocalBoxFile

Arguments:

key (FileKey, MainKey, ImportKey):: Decryption key. Must be specified if dlb argument is None.
dlb (DecryptedLocalBox, optional):: Decrypted LocalBox. Must be specified if key argument is None.
erase_encrypted_metadata (bool, optional):: Will remove metadata from the parent EncryptedLocalBoxFile after decryption to save more RAM if True. You can call .init() method on it to load it again.

property defaults: DefaultsTableWrapper | RemoteBoxDefaults#: Will return DefaultsTableWrapper or RemoteBoxDefaults.

async delete(remove_empty_directories: bool | None = False) → None#

Will delete this file from your LocalBox. You can re-import it from RemoteBox with import_file.

remove_empty_directories (bool, optional):

If True, will remove orphaned directories that left after removing files (if any).

Alternative: dlb.remove_empty_directories()

Note

This will delete file only from your LocalBox. To completely remove your file use same function on EncryptedRemoteBoxFile.

property directory: EncryptedLocalBoxDirectory | DecryptedLocalBoxDirectory | None#: Returns None i you call it on EncryptedLocalBoxFile that wasn’t initialized, EncryptedLocalBoxDirectory if on initialized or DecryptedLocalBoxDirectory if you call it on DecryptedLocalBoxFile

property dirkey: DirectoryKey | None#: Returns DirectoryKey of this file if present.

disable_cache_preview() → None#: Sets self._cache_preview to False and removes cached preview from memory.

property download_path: Path#: Returns current download path

property duration: int#: Returns media file duration.

enable_cache_preview() → None#: Sets self._cache_preview to True. Preview will be cached after first object.preview call.

property file_iv: bytes | None#: Returns file IV or None if file wasn’t initialized

property file_name: str#: Returns file name.

property file_path: Path#: Returns file path.

property file_salt: FileSalt | None#: Returns file salt or None if file wasn’t initialized

property filekey: FileKey#: Returns FileKey of this file.

property fingerprint: bytes | None#: Returns file fingerprint (hash of file path plus mainkey) or None

get_requestkey(mainkey: MainKey) → RequestKey#

Returns RequestKey for this File. You should use this method if you want to decrypt other’s EncryptedLocalBoxFile.

Arguments:

mainkey (MainKey):: To make a RequestKey for other’s EncryptedLocalBoxFile you need to have your own Box. Take key from it and specify here.

get_sharekey(reqkey: RequestKey | None = None) → ShareKey[source]#

Returns ShareKey for this file. You should use this method if you want to share your DecryptedLocalBoxFile with other people.

Arguments:

reqkey (RequestKey, optional):: Requester’s RequestKey. If isn’t specified returns ShareKey of this file without encryption, so ANYONE with this key can decrypt this local & remote box file.

property has_hmac_sha256: bool#: Will return True if file has HMAC (v1.5+)

property hmackey: HMACKey | None#: Returns HMACKey of this file if present.

property id: int | None#: Returns file ID or None if file wasn’t initialized

property imported: bool | None#: Returns True if file was forwarded to your BoxChannel.

async static init() → NoReturn[source]#: Will fetch and parse data from the Database.

property initialized: bool#: Returns True if you already called .init()

property is_encrypted: bool#: Will return True if this is an Encrypted class, False if Decrypted

property lb: EncryptedLocalBox | DecryptedLocalBox#: Will return EncryptedLocalBox from the EncryptedLocalBoxFile and DecryptedLocalBox from the DecryptedLocalBoxFile object.

property mime: bytes | None#: Returns mime type of the file

property minor_version: int | None#: Returns Minor Version of this file or None if class wasn’t initialized. If it’s a -1, then file was uploaded before the version 1.3.0 and minor is unknown.

property prefix: bytes | None#: Returns file prefix or None if file wasn’t initialized

property preview: bytes | None#: Returns preview bytes or b'' if cache_preview is False.

property residual_metadata: dict#: Will return metadata that left after parsing secret_metadata. This can be useful in future, when lower version will read file of a higher version.

set_download_path(path: Path)[source]#: Will set download path to specified.

property size: int#: Returns file size (no metadata included).

async update_metadata(changes: Dict[str, bytes | None], dlb: 'DecryptedLocalBox' | None = None, drb: 'DecryptedRemoteBox' | None = None, drbf: 'DecryptedRemoteBoxFile' | None = None)[source]#

This method will “update” file metadata attributes

In most cases you will want to update metadata in your Box as whole: in Remote & Local. You can specify here drbf (or at least drb) and we will do it automatically. However, you may want to update file metadata in the LocalBox only, and left the RemoteBox untouched. For such case do not specify drb or drbf.

Arguments:

changes (Dict[str, Union[bytes, None]]):

Metadata changes. You can specify a None as value to remove key from updates.

You can change the next fields: ‘duration’, ‘file_name’, ‘cattrs’, ‘mime’, ‘preview’ & ‘file_path’.

All values must be bytes. Use the tgbox.tools.int_to_bytes function for ‘duration’ field.

cattrs value should be packed with the tgbox.tools.PackedAttributes.

dlb (DecryptedLocalBox, optional):

If current local file wasn’t decrypted with the DecryptedLocalBox/MainKey then we can’t decrypt the efile_path (the new file_path) if it’s present. You can specify a DecryptedLocalBox to fix this. You don’t need to worry about this if you receive files from the DecryptedLocalBox

drb (DecryptedRemoteBox, optional):

DecryptedRemoteBox associated with this DecryptedLocalBox. Will auto refresh your updates in remote. Don’t specify this if you want to update metadata in the LocalBox only.

If you have DecryptedRemoteBoxFile, pass it as drbf instead.

drbf (DecryptedRemoteBoxFile, optional):

DecryptedRemoteBoxFile associated with this DecryptedLocalBoxFile. Will auto refresh your updates in remote. Don’t specify this if you want to update metadata in the LocalBox only.

E.g: This code will replace file_name metadata attribute of the DecryptedLocalBoxFile

... # Most code is omitted, see help(tgbox.api)
lfid = await dlb.get_last_file_id()
dlbf = await dlb.get_file(lfid)
await dlbf.update_metadata({'file_name': b'new.txt'})

print(dlbf.file_name) # new.txt

Note

Your RemoteBox will NOT know about this update, so you should specify here drb.
Not a default metadata (default is file_name, mime, etc) will be placed to the residual_metadata property dict.
LocalBox doesn’t have any limit on the CAttrs size except METADATA_MAX, but in RemoteBox there is a file caption (and so updated metadata) limit: 1KB and 2KB for a Premium Telegram users. Don’t specify drb if you want to update LocalBox only.
You can replace file’s path by specifying a file_path key with appropriate path (str/bytes). file_path='' will restore original file path. This is valid for all changed attributes.

property upload_time: bytes | int | None#: Returns encrypted upload_time from EncryptedLocalBoxFile and decrypted from DecryptedLocalBoxFile. None if class wasn’t initialized.

property version_byte: bytes | None#: Returns Verbyte of this file or None if class wasn’t initialized

class tgbox.api.local.EncryptedLocalBox(tgbox_db: TgboxDB, defaults_: DefaultsTableWrapper | RemoteBoxDefaults | None = None)[source]#

Bases: object

This class represents an encrypted local box. On more low-level that’s a wrapper around TgboxDB. Usually you will never meet this class in your typical code, but you may want to extract some encrypted data.

Usage:

from asyncio import run as asyncio_run
from tgbox.api import EncryptedLocalBox
from tgbox.db import TgboxDB

async def main():
    # Make or open TgboxDB
    tdb = await TgboxDB.create('TGBOX')
    # Initialize EncryptedLocalBox
    elb = await EncryptedLocalBox(tdb).init()

    # Retrieve encrypted session or None
    print(elb.session)

asyncio_run(main())

You can access it from the DecryptedLocalBox:

from asyncio import run as asyncio_run
from tgbox.api import get_localbox
from tgbox.keys import make_basekey, Phrase

phrase = Phrase(b'example phrase here')
basekey = make_basekey(phrase)

async def main():
    dlb = await get_localbox(basekey)
    # Retrieve encrypted session
    print(dlb._elb.session)

asyncio_run(main())

property api_hash: str | None#: Returns API_HASH.

property api_id: int | None#: Returns API_ID.

property box_channel_id: bytes | int | None#: Returns encrypted channel ID from EncryptedLocalBox and decrypted from DecryptedLocalBox.

property box_cr_time: bytes | int | None#: Returns encrypted box creation time from EncryptedLocalBox and decrypted from DecryptedLocalBox.

property box_salt: BoxSalt | None#: Returns BoxSalt or None if not initialized

Recursive iterate over all files/folders in LocalBox.

Arguments:

sfpid (bytes, optional):: Will start from this PartID if specified, else will start from “root” PIDs.
ignore_files (bool, optional):: Will not return LocalBoxFile associated with the LocalBoxDirectory if False.
erase_encrypted_metadata (bool, optional):: Will remove metadata from the parent EncryptedLocalBoxFile after decryption to save more RAM if True. You can call .init() method on it to load it again.

async decrypt(key: BaseKey | MainKey) → DecryptedLocalBox[source]#

Will return DecryptedLocalBox.

property defaults: DefaultsTableWrapper | RemoteBoxDefaults#: Returns DefaultsTableWrapper or RemoteBoxDefaults.

async delete() → None[source]#

This method WILL DELETE your LocalBox database. It doesn’t affect RemoteBox, so you can make new LocalBox from the Remote version if you have MainKey.

Will raise FileNotFoundError if something goes wrong (i.e DB was moved).

A function to remove a bunch of local files at once.

Arguments:

lbf (EncryptedLocalBoxFile, DecryptedLocalBoxFile, asterisk):

(Encrypted|Decrypted)LocalBoxFile(s) to remove.

lbf_ids (list, optional):

You can specify ids instead of LocalBox file objects. However, lbf is preferred here.

rb (EncryptedRemoteBox, DecryptedRemoteBox, optional):

You can specify a RemoteBox associated with current LocalBox to also remove all specified files in RemoteBox too.

remove_empty_directories (bool, optional):

If True, will remove orphaned directories that left after removing files (if any).

Alternative: dlb.remove_empty_directories()

Note

Without rb this will delete files only from your LocalBox. To completely remove your file use the same coroutine on RemoteBox or specify (Encrypted|Decrypted)RemoteBox.

async done()[source]#: Await this method when you end all work with LocalBox, so we will clean up & close connections.

Yields every local file as EncryptedLocalBoxFile if you call it on EncryptedLocalBox and DecryptedLocalBoxFile if on DecryptedLocalBox. Works via self.get_file

Alternatively, you may use LocalBox.contents method.

Arguments:

cache_preview (bool, optional):: Cache preview in class or not.
min_id (int, optional):: Will iterate from this ID.
max_id (int, optional):: Will iterate up to this ID.
ids (int, list, optional):: ID or list with IDs you want to fetch. If specified, The min_id and max_id args will be ignored
decrypt (bool, optional):: Will return EncryptedLocalBoxFile if False, and DecryptedLocalBoxFile if True. If None, will be determined by class.
reverse (bool, optional):: If set to True, the local files will be returned in reverse order (from newest to oldest, instead of the default oldest to newest).
fetch_count (int, optional):: Amount of files generator will fetch and cache from SQLite table before return. 100 by default.
erase_encrypted_metadata (bool, optional):: Will remove metadata from the parent EncryptedLocalBoxFile after decryption to save more RAM if True. You can call .init() method on it to load it again.

Returns EncryptedLocalBoxFile from EncryptedLocalBox or DecryptedLocalBoxFile from DecryptedLocalBox if file exists. None otherwise.

Arguments:

id (int, optional):: File ID. Must be specified if fingerprint is None.
fingerprint (bytes, optional):: File Fingerprint. Must be specified if id argument is None.
decrypt (bool, optional):: Will return EncryptedLocalBoxFile if False, and DecryptedLocalBoxFile if True. If None, will be determined by class.
cache_preview (bool, optional):: Cache preview in class or not.
erase_encrypted_metadata (bool, optional):: Will remove metadata from the parent EncryptedLocalBoxFile after decryption to save more RAM if True. You can call .init() method on it to load it again.

async get_files_total() → int[source]#: Returns a total number of files in this LocalBox

async get_last_file_id() → int[source]#: Returns last file id from the FILES table. If there is no files at all, will return 0.

get_requestkey(basekey: BaseKey) → RequestKey[source]#

Returns RequestKey for this LocalBox. You should use this method if you want to decrypt other’s RemoteBox.

Arguments:

basekey (BaseKey):: To make a RequestKey for other’s RemoteBox you need to create new BaseKey for it. Later this key will be used for Box decryption.

async init() → EncryptedLocalBox[source]#: Will fetch and parse data from Database.

property initialized: bool#: Returns True if you called .init()

property is_encrypted: bool#: Will return True if this is an Encrypted class, False if Decrypted

async remove_empty_directories(part_ids: List[bytes] | None = None)[source]#

By default, the Protocol will not automatically remove empty Directories after the file deletion process was done (unless forced). To do this, use this method.

Arguments:

part_ids (List[bytes], optional):

List of Part ID you want to check. Will remove only if Directory attached to this Part ID is orphaned (no files/other Directories linked)

If not specified, will check every Part ID in LocalBox.

property session: bytes | str | None#: Returns encrypted session from EncryptedLocalBox and decrypted from DecryptedLocalBox.

property tgbox_db: TgboxDB#: Returns TgboxDB.

class tgbox.api.local.EncryptedLocalBoxDirectory(elb: EncryptedLocalBox, part_id: bytes)[source]#

Bases: object

Class that represents abstract tgbox directory. You can iterate over all files/folders in it, as well as load parent folder up to root.

Typical usage:

from asyncio import run as asyncio_run
from tgbox.api import get_localbox
from tgbox.keys import make_basekey, Phrase

phrase = Phrase(b'example phrase here')
basekey = make_basekey(phrase)

async def main():
    dlb = await get_localbox(basekey)
    dlbfi = await dlb.get_file(await dlb.get_last_file_id())

    # Load directory up to the root
    await dlbfi.directory.lload(full=True)

    print(dlbfi.directory)

    # Iterate over files/folders in this directory
    async for content in dlbfi.directory.iterdir():
        print(content) # May be file or another dir

asyncio_run(main())

async decrypt(key: BaseKey | MainKey | None = None, dlb: DecryptedLocalBox | None = None)[source]#

Decrypt self and return DecryptedLocalBoxDirectory

Arguments:

key (BaseKey, MainKey, optional):: Decryption key. Must be specified if dlb is None.
dlb (DecryptedLocalBox, optional):: DecryptedLocalBox that we will use to decrypt EncryptedLocalBoxDirectory. Must be specified if key argument is None.

async delete() → None[source]#: Will delete this directory with all sub-dirs and files from your LocalBox. All of them will stay in RemoteBox, so you can restore all your data by syncing Box.

property floaded: bool#: Will return True if path was fully loaded

async get_contents_total() → int[source]#: Will return a total number of contents in this directory

async get_files_total() → int[source]#: Will return a total number of files in this directory

async get_folders_total() → int[source]#: Will return a total number of folders in this directory

async init() → EncryptedLocalBoxDirectory[source]#: Will fetch required data from the database.

property initialized: bool#: Returns True if you already called .init()

property is_encrypted: bool#: Will return True if this is an Encrypted class, False if Decrypted

Iterate over all files/folders in this abstract directory.

Arguments:

ignore_dirs (bool, optional):: Return abstract folders from this directory as LocalBoxDirectory or not. If False will return only LocalBoxFile objects.
ignore_files (bool, optional):: Will return LocalBoxFile if False.
cache_preview (bool, optional):: Cache preview in class or not. True by default.
ppid (bytes, DirectoryRoot):: Path PartID to iterate in. Will iterate over absolute LocalBox directory root if it’s DirectoryRoot. Will use self.part_id if not specified (by default).
erase_encrypted_metadata (bool, optional):: Will remove metadata from the parent EncryptedLocalBoxFile after decryption to save more RAM if True. You can call .init() method on it to load it again.

async lload(full: bool | None = False)[source]#

Will load and return one path part from the left side (previous) or return None if there is nothing left.

Arguments:

full (bool, optional):: If full is True, will load all parts from the left side and return None.

property parent_part_id: bytes | None#: Returns None if you didn’t called the .init() method, parent path part ID otherwise.

property part: bytes | None#: Returns None if you didn’t called the .init() method, encrypted path part if you’re in EncryptedLocalBoxDirectory, and decrypted if DecryptedLocalBoxDirectory.

property part_id: bytes#: Returns path part ID

property parts: list#: Will return loaded path parts.

property tgbox_db: TgboxDB#: Will return TgboxDB associated with this directory

class tgbox.api.local.EncryptedLocalBoxFile(id: int, elb: EncryptedLocalBox, cache_preview: bool = True)[source]#

Bases: object

This class represents an encrypted local file. On more low-level that’s a wrapper around row of FILES table in Tgbox Database. Usually you will not use this in your code.

from asyncio import run as asyncio_run
from tgbox.api import get_localbox
from tgbox.keys import make_basekey, Phrase

phrase = Phrase(b'example phrase here')
basekey = make_basekey(phrase)

async def main():
    dlb = await get_localbox(basekey)

    elbf = await dlb.get_file(
        id = await dlb.get_last_file_id(),
        decrypt = False
    )
    print(elbf.file_salt.hex())
    print(elbf.box_salt.hex())

asyncio_run(main())

property box_salt: BoxSalt | None#: Returns box salt or None if file wasn’t initialized

async decrypt(key: FileKey | ImportKey | MainKey | None = None, dlb: DecryptedLocalBox | None = None, erase_encrypted_metadata: bool = True) → DecryptedLocalBoxFile[source]#

Returns decrypted by key/dlb EncryptedLocalBoxFile

Arguments:

key (FileKey, MainKey, ImportKey):: Decryption key. Must be specified if dlb argument is None.
dlb (DecryptedLocalBox, optional):: Decrypted LocalBox. Must be specified if key argument is None.
erase_encrypted_metadata (bool, optional):: Will remove metadata from the parent EncryptedLocalBoxFile after decryption to save more RAM if True. You can call .init() method on it to load it again.

property defaults: DefaultsTableWrapper | RemoteBoxDefaults#: Will return DefaultsTableWrapper or RemoteBoxDefaults.

async delete(remove_empty_directories: bool | None = False) → None[source]#

Will delete this file from your LocalBox. You can re-import it from RemoteBox with import_file.

remove_empty_directories (bool, optional):

If True, will remove orphaned directories that left after removing files (if any).

Alternative: dlb.remove_empty_directories()

Note

This will delete file only from your LocalBox. To completely remove your file use same function on EncryptedRemoteBoxFile.

property directory: EncryptedLocalBoxDirectory | DecryptedLocalBoxDirectory | None#: Returns None i you call it on EncryptedLocalBoxFile that wasn’t initialized, EncryptedLocalBoxDirectory if on initialized or DecryptedLocalBoxDirectory if you call it on DecryptedLocalBoxFile

disable_cache_preview() → None[source]#: Sets self._cache_preview to False and removes cached preview from memory.

enable_cache_preview() → None[source]#: Sets self._cache_preview to True. Preview will be cached after first object.preview call.

property file_iv: bytes | None#: Returns file IV or None if file wasn’t initialized

property file_salt: FileSalt | None#: Returns file salt or None if file wasn’t initialized

property fingerprint: bytes | None#: Returns file fingerprint (hash of file path plus mainkey) or None

get_requestkey(mainkey: MainKey) → RequestKey[source]#

Returns RequestKey for this File. You should use this method if you want to decrypt other’s EncryptedLocalBoxFile.

Arguments:

mainkey (MainKey):: To make a RequestKey for other’s EncryptedLocalBoxFile you need to have your own Box. Take key from it and specify here.

property id: int | None#: Returns file ID or None if file wasn’t initialized

property imported: bool | None#: Returns True if file was forwarded to your BoxChannel.

async init() → EncryptedLocalBoxFile[source]#: Will fetch and parse data from the Database.

property initialized: bool#: Returns True if you already called .init()

property is_encrypted: bool#: Will return True if this is an Encrypted class, False if Decrypted

property lb: EncryptedLocalBox | DecryptedLocalBox#: Will return EncryptedLocalBox from the EncryptedLocalBoxFile and DecryptedLocalBox from the DecryptedLocalBoxFile object.

property minor_version: int | None#: Returns Minor Version of this file or None if class wasn’t initialized. If it’s a -1, then file was uploaded before the version 1.3.0 and minor is unknown.

property prefix: bytes | None#: Returns file prefix or None if file wasn’t initialized

property upload_time: bytes | int | None#: Returns encrypted upload_time from EncryptedLocalBoxFile and decrypted from DecryptedLocalBoxFile. None if class wasn’t initialized.

property version_byte: bytes | None#: Returns Verbyte of this file or None if class wasn’t initialized

async tgbox.api.local.clone_remotebox(drb: tgbox.api.remote.DecryptedRemoteBox, basekey: BaseKey, progress_callback: Callable[[int, int], None] | None = None, box_name: str | None = None, box_path: PathLike | str | None = None, timeout: int | None = 15) → DecryptedLocalBox[source]#

This method makes LocalBox from RemoteBox and imports all RemoteBoxFiles to it.

Arguments:

drb (DecryptedRemoteBox):: DecryptedRemoteBox you want to clone.
basekey (BaseKey):: BaseKey with which you will decrypt your cloned EncryptedLocalBox. BaseKey encrypts Session and MainKey of original LocalBox.
progress_callback (Callable[[int, int], None], optional):: A callback function accepting two parameters: (current_amount, total).
box_name (str, optional):: Filename of your LocalBox database. If not specified, will be used RemoteBox name.
box_path (PathLike, str, optional):: Path in which we will make a database file. Current Working Dir if not specified.
timeout (int, optional):: How many seconds generator will sleep at every 1000 file. By default it’s 15 seconds. Don’t use too low timeouts or you will receive FloodWaitError.

async tgbox.api.local.get_localbox(basekey: BaseKey | None = None, tgbox_db_path: PathLike | str | None = None) → EncryptedLocalBox | DecryptedLocalBox[source]#

Returns LocalBox.

Arguments:

basekey (BaseKey, optional):: Returns DecryptedLocalBox if specified, EncryptedLocalBox otherwise (default).
tgbox_db_path (PathLike, str, optional):: PathLike to your TgboxDB (LocalBox). Default is defaults.DEF_TGBOX_NAME.

async tgbox.api.local.make_localbox(erb: tgbox.api.remote.EncryptedRemoteBox, basekey: BaseKey, box_name: str | None = None, box_path: PathLike | str | None = None) → DecryptedLocalBox[source]#

Makes LocalBox

Arguments:

erb (RemoteBox):: EncryptedRemoteBox. You will receive it after make_remotebox.
basekey (BaseKey):: BaseKey that will be used for MainKey creation.
box_name (str, optional):: Filename of your LocalBox database. If not specified, will be used RemoteBox name.
box_path (PathLike, str, optional):: Path in which we will make a database file. Current Working Dir if not specified.

tgbox.api.remote module#

Module with API functions and classes for RemoteBox.

class tgbox.api.remote.DecryptedRemoteBox(erb: EncryptedRemoteBox, key: MainKey | ImportKey | None = None, dlb: DecryptedLocalBox | None = None)[source]#

Bases: EncryptedRemoteBox

RemoteBox is a remote cloud storage. You can upload files and download them later.

Locally we only keep info about files (in LocalBox). You can fully restore your LocalBox from RemoteBox.

This class represents decrypted RemoteBox, you can iterate over all decrypted files, clone and upload.

from asyncio import run as asyncio_run
from tgbox.api import get_localbox, get_remotebox
from tgbox.keys import make_basekey, Phrase

phrase = Phrase(b'example phrase here')
basekey = make_basekey(phrase)

async def main():
    dlb = await dlb.get_localbox(basekey)
    drb = await get_remotebox(dlb)

    # Make a PreparedFile
    pf = await dlb.prepare_file(open('cats.jpg','rb'))

    # Waiting file for upload, return DecryptedRemoteBoxFile
    drbf = await drb.push_file(pf)

    # Get some info
    print(drbf.file_name, drbf.size)

    # Remove file from RemoteBox
    await drbf.delete()

    # Check if file exists
    print(await drb.file_exists(drbf.id)

asyncio_run(main())

async author_files(toggle: bool) → bool | None#

This function will enable or disable (by toggle) file(/message) authoring.

Same as Box Channel Settings -> “Sign Messages” -> “Show author’s profiles”

Arguments:

toggle (bool):: If True, will enable authoring. If False, will disable it.

Will return True if file authoring was enabled or False if you don’t have enough privileges to change it.

property box_channel: Channel#: Returns instance of Channel

property box_channel_id: int#: Returns box channel id

async static decrypt() → NoReturn[source]#

property defaults: DefaultsTableWrapper | RemoteBoxDefaults#: Will return DefaultsTableWrapper if dlb was specified or RemoteBoxDefaults with arguments from the defaults module if wasn’t.

async delete() → None#

This method WILL DELETE RemoteBox!

Use left() if you only want to left your Box Channel, not delete it.

You need to have admin rights for this.

async delete_files(*rbf: EncryptedRemoteBoxFile | DecryptedRemoteBoxFile, rbf_ids: list | None = None, lb: tgbox.api.local.EncryptedLocalBox | tgbox.api.local.DecryptedLocalBox | None = None) → None#

A function to remove a bunch of remote files at once. You need to have some admin rights.

Arguments:

rbf (EncryptedRemoteBoxFile, DecryptedRemoteBoxFile, asterisk):: (Encrypted|Decrypted)RemoteBoxFile(s) to remove.
rbf_ids (list, optional):: You can specify ids instead of RemoteBox file objects. However, rbf is preferred here.
lb (EncryptedLocalBox, DecryptedLocalBox, optional):: You can specify a LocalBox associated with current RemoteBox to also remove all specified files in LocalBox too.

Note

If you want to delete files only from your LocalBox then you can use the same method on your LocalBoxFile.

async done()#: Await this method when you end all work with RemoteBox, so we will clean up & close connections.

async file_exists(id: int) → bool#

Returns True if file with specified id exists in RemoteBox. False otherwise.

Arguments:

id (int):: File ID.

async files(key: MainKey | FileKey | None = None, drb: DecryptedRemoteBox | None = None, dlb: DecryptedLocalBox | None = None, *, ignore_errors: bool = True, return_imported_as_erbf: bool = False, limit: int | None = None, offset_id: int = 0, max_id: int = 0, min_id: int = 0, add_offset: int = 0, search: str | None = None, from_user: int | str | None = None, wait_time: float | None = None, ids: int | List[int] | None = None, reverse: bool = False, decrypt: bool | None = None, timeout: int = 15, cache_preview: bool = True, erase_encrypted_metadata: bool = True) → AsyncGenerator[EncryptedRemoteBoxFile | DecryptedRemoteBoxFile, None]#

Yields every RemoteBoxFile from RemoteBox.

Note

The default order is from newest to oldest, but this behaviour can be changed with the reverse parameter.
You can ignore key and drb if you call this method on DecryptedRemoteBox.

Arguments:

key (MainKey, FileKey, optional):

Will be used to decrypt EncryptedRemoteBoxFile.

drb (DecryptedRemoteBox):

Decrypted RemoteBox. Will be used to decrypt EncryptedRemoteBoxFile

dlb (DecryptedLocalBox, optional):

If file in your RemoteBox was imported from other RemoteBox, then you can’t decrypt it with specified mainkey, but if you already imported it to your LocalBox, then you can specify dlb and we will use FILEKEY from the Database.

If decrypt specified but there is no key, then we will try to use mainkey from this dlb.

This kwarg works in tandem with ignore_errors and return_imported_as_erbf if dlb doesn’t have this file (tip: you need to import it with dlb.import_file.

ignore_errors (bool, optional):

Ignore all errors related to decryption of the files in your RemoteBox. If True, (by default) only yields files that was successfully decrypted. Can be useful if you have files that was imported from other RemoteBox and you don’t want to specify dlb.

return_imported_as_erbf (bool, optional):

If specified, yields files that generator can’t decrypt (imported) as EncryptedRemoteBoxFile.

limit (int | None, optional):

Number of files to be retrieved. Due to limitations with the API retrieving more than 3000 messages will take longer than half a minute (or even more based on previous calls). The limit may also be None, which would eventually return the whole history.

offset_id (int, optional):

Offset message ID (only remote files previous to the given ID will be retrieved). Exclusive.

max_id (int, optional):

All the remote box files with a higher (newer) ID or equal to this will be excluded.

min_id (int, optional):

All the remote box files with a lower (older) ID or equal to this will be excluded.

add_offset (int, optional):

Additional message offset (all of the specified offsets + this offset = older files).

search (str, optional):

The string to be used as a search query.

from_user (str, int, optional):

Only messages from this entity will be returned.

wait_time (int, optional):

Wait time (in seconds) between different GetHistoryRequest (Telethon). Use this parameter to avoid hitting the ``FloodWaitError`` as needed. If left to None, it will default to 1 second only if the limit is higher than 3000. If the ``ids`` parameter is used, this time will default to 10 seconds only if the amount of IDs is higher than 300.

ids (int, list, optional):

A single integer ID (or several IDs) for the box files that should be returned. This parameter takes precedence over the rest (which will be ignored if this is set). This can for instance be used to get the file with ID 123 from a box channel. Note that if the file-message doesn’t exist, None will appear in its place, so that zipping the list of IDs with the files can match one-to-one.

reverse (bool, optional):

If set to True, the remote files will be returned in reverse order (from newest to oldest, instead of the default oldest to newest). This also means that the meaning of offset_id parameter is reversed, although offset_id still be exclusive. min_id becomes equivalent to offset_id instead of being max_id as well since files are returned in ascending order.

timeout (int, optional):

How many seconds generator will sleep at every 1000 file. By default it’s 15 seconds. Don’t use too low timeouts, you will receive FloodWaitError otherwise (TGBOX).

decrypt (bool, optional):

Returns DecryptedRemoteBoxFile if True, EncryptedRemoteBoxFile otherwise. If None, will be determined by class.

cache_preview (bool, optional):

Cache preview in yielded by generator RemoteBoxFiles or not. True by default.

erase_encrypted_metadata (bool, optional):

Will remove metadata from the parent EncryptedRemoteBoxFile after decryption to save more RAM if True. You can call .init() method on it to load it again.

async get_box_description(force: bool | None = False)#: Returns Box description if presented. If force specified, will make request & update box_salt and description.

async get_box_name()#: Returns name of RemoteBox. Will be cached after first method call.

async get_box_salt(force: bool | None = False) → BoxSalt#: Returns BoxSalt. Will be cached after first method call. If force specified, will make request & update box_salt & description

async get_file(id: int, key: MainKey | FileKey | ImportKey | None = None, dlb: DecryptedLocalBox | None = None, decrypt: bool | None = None, ignore_errors: bool = True, return_imported_as_erbf: bool = False, cache_preview: bool = True, erase_encrypted_metadata: bool = True) → EncryptedRemoteBoxFile | DecryptedRemoteBoxFile | None#

Returns file from the RemoteBox by the given ID.

Note

You may ignore key` and ``dlb if you call this method on DecryptedRemoteBox.

Arguments:

id (int):

File ID.

key (MainKey, FileKey, optional):

Will be used to decrypt EncryptedRemoteBoxFile.

dlb (DecryptedLocalBox, optional):

If file in your RemoteBox was imported from other RemoteBox then you can’t decrypt it with specified mainkey, but if you already imported it to your LocalBox, then you can specify dlb and we will use FILE_KEY from the Database.

If decrypt specified but there is no key, then we try to use mainkey from this dlb.

This kwarg works in tandem with ignore_errors and return_imported_as_erbf if dlb doesn’t have this file (tip: you need to import it with dlb.import_file).

decrypt (bool, optional):

Returns DecryptedRemoteBoxFile if True, EncryptedRemoteBoxFile otherwise. If None, will be determined by class.

ignore_errors (bool, optional):

Ignore all errors related to decryption of the files in your RemoteBox. If True, (by default) only returns file that was successfully decrypted. Can be useful if you have files that was imported from other RemoteBox and you don’t want to specify dlb.

return_imported_as_erbf (bool, optional):

If specified, returns file that method can’t decrypt (if imported) as EncryptedRemoteBoxFile.

cache_preview (bool, optional):

Cache preview in returned by method RemoteBoxFiles or not. True by default.

erase_encrypted_metadata (bool, optional):

Will remove metadata from the parent EncryptedRemoteBoxFile after decryption to save more RAM if True. You can call .init() method on it to load it again.

async get_files_total() → int#: Returns a total number of files in this RemoteBox

async get_last_file_id() → int#: Returns last channel file id. If nothing found returns 0

async get_requestkey(basekey: BaseKey) → RequestKey#

Returns RequestKey for this RemoteBox. You should use this method if you want to decrypt other’s RemoteBox.

Arguments:

basekey (BaseKey):: To make a RequestKey for other’s RemoteBox you need to create new BaseKey for it. Later this key will be used for Box decryption.

async get_sharekey(reqkey: RequestKey | None = None) → ShareKey[source]#

Returns ShareKey for this Box. You should use this method if you want to share your RemoteBox with other people.

Arguments:

reqkey (RequestKey, optional):: Requester’s RequestKey. If isn’t specified returns ShareKey of this box without encryption, so anyone with this key can decrypt ALL files in your RemoteBox.

property is_encrypted: bool#: Will return True if this is an Encrypted class, False if Decrypted

async left() → None#: With calling this method you will left RemoteBox Channel.

property mainkey: MainKey#: Will return MainKey of this Box

async push_file(pf: PreparedFile, progress_callback: Callable[[int, int], None] | None = None, use_slow_upload: bool | None = False) → DecryptedRemoteBoxFile#

Uploads PreparedFile to the RemoteBox.

Arguments:

pf (PreparedFile):: PreparedFile to upload. You should receive it via DecryptedLocalBox.prepare_file.
progress_callback (Callable[[int, int], None], optional):: A callback function accepting two parameters: (downloaded_bytes, total).
use_slow_upload (bool, optional):: Will use default upload function from the Telethon library instead of function from fastelethon.py. Use this if you have problems with upload.

async search_file(sf: SearchFilter, mainkey: MainKey | None = None, dlb: DecryptedLocalBox | None = None, cache_preview: bool = True, return_imported_as_erbf: bool = False, reverse: bool = False) → AsyncGenerator[EncryptedRemoteBoxFile | DecryptedRemoteBoxFile, None]#

This method used to search for files in your RemoteBox.

Arguments:

sf (SearchFilter):: SearchFilter with kwargs you like.
mainkey (MainKey, optional):: MainKey for this RemoteBox.
dlb (DecryptedLocalBox, optional):: LocalBox associated with this RemoteBox. We will take MainKey from it.
cache_preview (bytes, optional):: Will cache preview in file object if True.
return_imported_as_erbf (bool, optional):: If specified, will yield files that generator can’t decrypt (imported) as EncryptedRemoteBoxFile.
reverse (bool, optional):: If set to True, the remote files will be returned in reverse order (from newest to oldest, instead of the default oldest to newest).

Note

If dlb and mainkey not specified, then method will search only for EncryptedRemoteBoxFile.
You may ignore this kwargs if you call this method on DecryptedRemoteBox class.

async sign_files(toggle: bool) → bool#

This function will enable or disable (by toggle) file(/message) sender signature.

Same as Box Channel Settings -> “Sign Messages”

Arguments:

toggle (bool):: If True, will enable signatures. If False, will disable message signing.

Will return True if file signing was enabled or False if you don’t have enough privileges to change it.

property tc: TelegramClient#: Returns TelegramClient

async update_file(rbf: EncryptedRemoteBoxFile | DecryptedRemoteBoxFile, pf: PreparedFile, progress_callback: Callable[[int, int], None] | None = None, use_slow_upload: bool | None = False) → DecryptedRemoteBoxFile#

Updates already uploaded RemoteBox file. This will make a full reupload and Message edit.

rbf (EncryptedRemoteBoxFile, DecryptedRemoteBoxFile):: The RemoteBox file to update. We will only take a Message object from it. The rbf will NOT be updated by itself, instead, new RemoteBox file object will be returned after update.
pf (PreparedFile):: PreparedFile to upload. You should receive it via DecryptedLocalBox.prepare_file (set skip_fingerprint_check to True).
progress_callback (Callable[[int, int], None], optional):: A callback function accepting two parameters: (downloaded_bytes, total).
use_slow_upload (bool, optional):: Will use default upload function from the Telethon library instead of function from fastelethon.py. Use this if you have problems with upload.

class tgbox.api.remote.DecryptedRemoteBoxFile(erbf: EncryptedRemoteBoxFile, key: MainKey | FileKey | ImportKey | None = None, drb: DecryptedRemoteBox | None = None, cache_preview: bool | None = None, erase_encrypted_metadata: bool = True)[source]#

Bases: EncryptedRemoteBoxFile

This class represents decrypted remote file. You can retrieve all metadata info from properties.

Typical usage:

from asyncio import run as asyncio_run
from tgbox.api import get_localbox, get_remotebox
from tgbox.keys import Phrase, make_basekey

phrase = Phrase(b'example phrase here')
basekey = make_basekey(phrase)

async def main():
    dlb = await get_localbox(basekey)
    drb = await get_remotebox(dlb)

    drbf = await drb.get_file(
        id = await dlb.get_last_file_id()
    )
    print(drbf.file_name)

    # Save file preview
    with open(f'preview_{drbf.file_name}','wb') as f:
        f.write(drbf.preview)

    # Download file, return BinaryIO
    file = await drbf.download()

asyncio_run(main())

property box_channel: Channel#: Returns RemoteBox Channel object.

property box_channel_id: int#: Returns ID of the RemoteBox Channel.

property box_salt: BoxSalt | None#: Returns BoxSalt or None if not initialized

property cattrs: dict | None#: Returns custom attributes or None if not initialized

async static decrypt() → NoReturn[source]#

Returns DecryptedRemoteBoxFile.

Arguments:

key (FileKey, MainKey, ImportKey):: Decryption key. Must be specified if drb argument is None.
drb (DecryptedRemoteBox, optional):: Decrypted RemoteBox. Must be specified if key argument is None.
erase_encrypted_metadata (bool, optional):: Will remove metadata from the parent EncryptedRemoteBoxFile after decryption to save more RAM if True. You can call .init() method on it to load it again.

property defaults: DefaultsTableWrapper | RemoteBoxDefaults#: Will return DefaultsTableWrapper or RemoteBoxDefaults.

async delete() → None#: TOTALLY removes file from RemoteBox. You and all participants of the EncryptedRemoteBox will lose access to it FOREVER. This action can’t be undone. You need to have rights for this action.

Note

If you want to delete file only from your LocalBox then you can use the same delete() method on your LocalBoxFile.

property dirkey: DirectoryKey | None#: Returns DirectoryKey of this file if present.

async download(*, outfile: str | BinaryIO | Path | None = None, hide_folder: bool = False, hide_name: bool = False, decrypt: bool = True, request_size: int = 524288, offset: int | None = None, progress_callback: Callable[[int, int], None] | None = None, use_slow_download: bool | None = False, hmac_state: HMAC | None = None, omit_hmac_check: bool | None = False) → BinaryIO[source]#

Downloads and saves remote box file to the outfile.

Arguments:

oufile (str, BinaryIO, PathLike, optional):

Path-like or File-like object to which file will be downloaded. self.defaults.DOWNLOAD_PATH by default.

If outfile has .write() method then we will use it.

If outfile is str or Path and offset, – we will open in ab+ mode, wb otherwise.

If outfile is your custom object, then you need to implement a .write() at minimum. If you download from offset then also .read() and .seek() or pass hmac_state as keyword argument instead.

hide_folder (bool, optional):

Saves to folder which this file belongs to if False, (default) otherwise to outfile/{self.defaults.DEF_UNK_FOLDER}.

Doesn’t create any folders if isinstance(outfile, BinaryIO).

hide_name (bool, optional):

Saves file with random name if True, with original if False (default).

File extension (e.g .png) included in both cases.
Doesn’t create any folders if isinstance(outfile, BinaryIO).

decrypt (bool, optional):

Decrypts file if True (default).

offset (int, optional):

Offset to decrypted file. Use this if your download process was stopped for some reason. Specify here how much of bytes you already downloaded and we will fetch rest.

request_size (int, optional):

How many bytes will be requested to Telegram when more data is required. By default, as many bytes as possible are requested. If you would like to request data in smaller sizes, adjust this parameter.

Note that values outside the valid range will be clamped, and the final value will also be a multiple of the minimum allowed size.

progress_callback (Callable[[int, int], None], optional):

A callback function accepting two parameters: (downloaded_bytes, total).

use_slow_download (bool, optional):

Will use default download function from the Telethon library instead of function from fastelethon.py. Use this if you have problems with download.

hmac_state (hmac.HMAC, optional):

If you download file from some offset and your outfile is NOT readable, then we can’t compute and verify the HMAC checksum. In this case, you will need to provide a hmac.HMAC state with bytes updated up to the offset. For example:

... # Most code was omitted

from hmac import HMAC

# Let's assume that our download process was
# disrupted by some event. We already fetched
# big part of file, so we don't want to make a
# full re-download. Also, we assume that we
# download to custom 'outfile' which is NOT
# readable (for example, here we will open it
# in 'ab' mode, however, 'ab+' would be readable)

outfile = open('video.mp4','ab') # NOT readable!

# This is corresponding file that we want to download
drbf = await drb.get_file(dlb.get_last_file_id())

# Create 'hmac_state' and init with DRBF 'HMACKey'
hmac_state = HMAC(drbf.hmackey.key, digestmod='sha256')

# Now we need to update a 'hmac_state' with bytes
# that we already downloaded. AGAIN, this is a
# STUPID example. If you can make the 'outfile'
# readable (i.e 'ab+'), then just pass it to
# 'download()' method as is! OTHERWISE:

with open(outfile.name,'rb') as f:
    hmac_state.update(f.read()) # Update hmac_state
    offset = f.tell() # Easily retrieve offset

await drbf.download(
    outfile=outfile,
    offset=offset,
    hmac_state=hmac_state
)
# This is just example with 'open()'. If you
# have custom object as 'outfile' then you need
# to made it in different way suitable for you

omit_hmac_check (bool, optional):

Will omit HMAC check on download if True. As we make HMAC of plaintext on upload, HMAC check be always skipped if decrypt is False.

property duration: float | None#: Returns duration or None if not initialized.

property file: File#: Returns Telethon’s File object.

property file_file_name: bytes#: Returns remote file name.

property file_iv: IV | None#: Returns IV or None if not initialized.

property file_name: str | None#: Returns file name or None if not initialized.

property file_path: Path | None#: Returns file path or None if not initialized.

property file_salt: FileSalt | None#: Returns FileSalt or None if not initialized.

property file_size: int#: Returns size of the File from Message object.

property filekey: FileKey#: Returns FileKey of this file.

property fingerprint: bytes | None#: Returns file fingerprint (hash of file path plus mainkey) or None

async get_imported_from_entity() → Channel | User | None#

Returns forward author entity (the one Document was forwarded from, i.e User / Channel).

If Author is Channel, Channel object will be returned.
If Author is User, User object will be returned.

get_requestkey(mainkey: MainKey) → RequestKey#

Returns RequestKey for this file. You should use this method if you want to decrypt other’s EncryptedRemoteBoxFile.

Arguments:

mainkey (MainKey):: To make a RequestKey for other’s RemoteBoxFile you need to have your Box. Take key from your DecryptedLocalBox and specify it here.

async get_sender_entity() → Channel | User | None#

Returns post author entity if “Sign Messages” with “Show author’s profiles” was enabled in Box Channel, None otherwise.

If Author is Channel, Channel object will be returned.
If Author is User, User object will be returned.

get_sharekey(reqkey: RequestKey | None = None) → ShareKey[source]#

Returns ShareKey for this file. You should use this method if you want to share this file with other people.

Arguments:

reqkey (RequestKey, optional):: Requester’s RequestKey. If isn’t specified returns ImportKey of this file without encryption, so ANYONE with this key can decrypt this remote file.

property has_hmac_sha256: bool#: Will return True if file has HMAC (v1.5+)

property hmackey: HMACKey | None#: Returns HMACKey of this file if present.

property id: int#: Returns message id.

property imported: bool#: Returns True if file was imported from other RemoteBox. False otherwise.

property imported_from_id: int | None#

Returns forward author (the entity Document was forwarded from, i.e User / Channel) ID.

If Author is Channel, ID will be negative.
If User, ID will be always positive (> 0).

async static init() → NoReturn[source]#

This method will download and set raw RemoteBoxFile metadata. If metadata length is bigger than self.defaults.METADATA_MAX then errors.LimitExceeded will be raised.

Arguments:

verify_prefix (bool, optional):: If True, will check that file has a defaults.PREFIX in metadata, and if not, will raise a NotATgboxFile exception.

property initialized: bool#: Returns True if class was initialized.

property is_encrypted: bool#: Will return True if this is an Encrypted class, False if Decrypted

property message: Message#: Returns Telethon’s Message object.

property mime: str | None#: Returns MIME type or None if not initialized

property minor_version: int | None#: Returns Minor Version of this file or None if class wasn’t initialized. If it’s a -1, then file was uploaded before the version 1.3.0 and minor is unknown.

property prefix: bytes | None#: Returns file prefix or None if not initialized

property preview: bytes | None#: Returns preview bytes or b'' if not initialized.

property rb: EncryptedRemoteBox | DecryptedRemoteBox#: Returns EncryptedRemoteBox from EncryptedRemoteBoxFile and DecryptedRemoteBox from DecryptedRemoteBoxFile

property residual_metadata: dict | None#

Will return metadata that left after parsing secret_metadata. This can be useful in future, when lower version will read file of a higher version.

Will always return None if DRBFI wasn’t initialized.

property sender: str | None#: Returns post author if “Sign Messages” is enabled in Box Channel, None otherwise.

property sender_id: int | None#

Returns post author ID if “Sign Messages” with “Show author’s profiles” was enabled in Box Channel, None otherwise.

If Sender is Channel, ID will be negative.
If User, ID will be always positive (> 0).

set_file_path(file_path: Path) → None[source]#: Will change self._file_path to file_path. In most cases you don’t need to use this

property size: int | None#: Returns file size or None if not initialized.

async update_metadata(changes: Dict[str, bytes | None])[source]#

This method will “update” file metadata attributes

Metadata located inside the file, so we can’t change it in any way except reupload, but we can and we will use its caption to place packed by PackedAttributes, encrypted and encoded with urlsafe_b64encode changes.

This method will update metadata only in the RemoteBox. If you want to update metadata in whole Box (Remote & Local), then use the same method on the DecryptedLocalBoxFile and specify this DecryptedRemoteBoxFile as drb.

Arguments:

changes (Dict[str, Union[bytes, None]]):

Metadata changes. You can specify a None as value to remove key from updates.

You can change the next fields: ‘duration’, ‘file_name’, ‘cattrs’, ‘mime’, ‘preview’ & ‘file_path’.

All values must be bytes. Use the tgbox.tools.int_to_bytes function for ‘duration’ field.

E.g: This code will replace file_name metadata attribute of the DecryptedRemoteBoxFile

... # Most code is omitted, see help(tgbox.api)
lfid = await drb.get_last_file_id()
drbf = await drb.get_file(lfid)
await drbf.update_metadata({'file_name': b'new.txt'})

print(drbf.file_name) # new.txt

Note

Your LocalBox will NOT know about this update. If you want to update Remote & Local use same method on the DecryptedLocalBoxFile with drb.
Not a default metadata (default is file_name, mime, etc) will be placed to the residual_metadata property dict.
You can replace file’s path by specifying a file_path key with appropriate path (str/bytes). file_path='' will restore original file path. This is valid for all changed attributes.
There is a file caption (and so updated metadata) limit: 1KB and 2KB for a Premium Telegram users.

property updated_at_time: int | None#: Returns time when file was updated or None if not initialized

property upload_time: int | None#: Returns upload time or None if not initialized

property version_byte: bytes | None#: Returns Verbyte or None if not initialized

class tgbox.api.remote.EncryptedRemoteBox(box_channel: Channel, tc: TelegramClient, defaults_: RemoteBoxDefaults | DefaultsTableWrapper | None = None)[source]#

Bases: object

RemoteBox is a remote cloud storage. You can upload files and download them later.

Locally we only keep info about files (in LocalBox). You can fully restore your LocalBox from RemoteBox.

Note

In EncryptedRemoteBox you should specify MainKey or DecryptedLocalBox. Usually you want to use DecryptedRemoteBox, not this class.

Typical usage:

from tgbox.api import (
    TelegramClient,
    make_localbox,
    make_remotebox
)
from getpass import getpass
from asyncio import run as asyncio_run

PHONE_NUMBER = '+10000000000' # Your phone number
API_ID = 1234567 # Your own API_ID: my.telegram.org
API_HASH = '00000000000000000000000000000000' # Your own API_HASH

async def main():
    # Connect and sign-in to Telegram
    tc = TelegramClient(
        phone_number = PHONE_NUMBER,
        api_id = API_ID,
        api_hash = API_HASH
    )
    await tc.connect()
    await tc.send_code()

    await tc.log_in(
        code = int(input('Code: ')),
        password = getpass('Pass: ')
    )
    # Making base RemoteBox (EncryptedRemoteBox)
    erb = await make_remotebox(tc)

asyncio_run(main())

async author_files(toggle: bool) → bool | None[source]#

This function will enable or disable (by toggle) file(/message) authoring.

Same as Box Channel Settings -> “Sign Messages” -> “Show author’s profiles”

Arguments:

toggle (bool):: If True, will enable authoring. If False, will disable it.

Will return True if file authoring was enabled or False if you don’t have enough privileges to change it.

property box_channel: Channel#: Returns instance of Channel

property box_channel_id: int#: Returns box channel id

async decrypt(*, key: MainKey | ImportKey | BaseKey | None = None, dlb: DecryptedLocalBox | None = None) → DecryptedRemoteBox[source]#

property defaults: DefaultsTableWrapper | RemoteBoxDefaults#: Will return DefaultsTableWrapper if dlb was specified or RemoteBoxDefaults with arguments from the defaults module if wasn’t.

async delete() → None[source]#

This method WILL DELETE RemoteBox!

Use left() if you only want to left your Box Channel, not delete it.

You need to have admin rights for this.

A function to remove a bunch of remote files at once. You need to have some admin rights.

Arguments:

rbf (EncryptedRemoteBoxFile, DecryptedRemoteBoxFile, asterisk):: (Encrypted|Decrypted)RemoteBoxFile(s) to remove.
rbf_ids (list, optional):: You can specify ids instead of RemoteBox file objects. However, rbf is preferred here.
lb (EncryptedLocalBox, DecryptedLocalBox, optional):: You can specify a LocalBox associated with current RemoteBox to also remove all specified files in LocalBox too.

Note

If you want to delete files only from your LocalBox then you can use the same method on your LocalBoxFile.

async done()[source]#: Await this method when you end all work with RemoteBox, so we will clean up & close connections.

async file_exists(id: int) → bool[source]#

Returns True if file with specified id exists in RemoteBox. False otherwise.

Arguments:

id (int):: File ID.

Yields every RemoteBoxFile from RemoteBox.

Note

The default order is from newest to oldest, but this behaviour can be changed with the reverse parameter.
You can ignore key and drb if you call this method on DecryptedRemoteBox.

Arguments:

key (MainKey, FileKey, optional):

Will be used to decrypt EncryptedRemoteBoxFile.

drb (DecryptedRemoteBox):

Decrypted RemoteBox. Will be used to decrypt EncryptedRemoteBoxFile

dlb (DecryptedLocalBox, optional):

If decrypt specified but there is no key, then we will try to use mainkey from this dlb.

This kwarg works in tandem with ignore_errors and return_imported_as_erbf if dlb doesn’t have this file (tip: you need to import it with dlb.import_file.

ignore_errors (bool, optional):

return_imported_as_erbf (bool, optional):

If specified, yields files that generator can’t decrypt (imported) as EncryptedRemoteBoxFile.

limit (int | None, optional):

offset_id (int, optional):

Offset message ID (only remote files previous to the given ID will be retrieved). Exclusive.

max_id (int, optional):

All the remote box files with a higher (newer) ID or equal to this will be excluded.

min_id (int, optional):

All the remote box files with a lower (older) ID or equal to this will be excluded.

add_offset (int, optional):

Additional message offset (all of the specified offsets + this offset = older files).

search (str, optional):

The string to be used as a search query.

from_user (str, int, optional):

Only messages from this entity will be returned.

wait_time (int, optional):

ids (int, list, optional):

reverse (bool, optional):

timeout (int, optional):

How many seconds generator will sleep at every 1000 file. By default it’s 15 seconds. Don’t use too low timeouts, you will receive FloodWaitError otherwise (TGBOX).

decrypt (bool, optional):

Returns DecryptedRemoteBoxFile if True, EncryptedRemoteBoxFile otherwise. If None, will be determined by class.

cache_preview (bool, optional):

Cache preview in yielded by generator RemoteBoxFiles or not. True by default.

erase_encrypted_metadata (bool, optional):

Will remove metadata from the parent EncryptedRemoteBoxFile after decryption to save more RAM if True. You can call .init() method on it to load it again.

async get_box_description(force: bool | None = False)[source]#: Returns Box description if presented. If force specified, will make request & update box_salt and description.

async get_box_name()[source]#: Returns name of RemoteBox. Will be cached after first method call.

async get_box_salt(force: bool | None = False) → BoxSalt[source]#: Returns BoxSalt. Will be cached after first method call. If force specified, will make request & update box_salt & description

Returns file from the RemoteBox by the given ID.

Note

You may ignore key` and ``dlb if you call this method on DecryptedRemoteBox.

Arguments:

id (int):

File ID.

key (MainKey, FileKey, optional):

Will be used to decrypt EncryptedRemoteBoxFile.

dlb (DecryptedLocalBox, optional):

If decrypt specified but there is no key, then we try to use mainkey from this dlb.

This kwarg works in tandem with ignore_errors and return_imported_as_erbf if dlb doesn’t have this file (tip: you need to import it with dlb.import_file).

decrypt (bool, optional):

Returns DecryptedRemoteBoxFile if True, EncryptedRemoteBoxFile otherwise. If None, will be determined by class.

ignore_errors (bool, optional):

return_imported_as_erbf (bool, optional):

If specified, returns file that method can’t decrypt (if imported) as EncryptedRemoteBoxFile.

cache_preview (bool, optional):

Cache preview in returned by method RemoteBoxFiles or not. True by default.

erase_encrypted_metadata (bool, optional):

Will remove metadata from the parent EncryptedRemoteBoxFile after decryption to save more RAM if True. You can call .init() method on it to load it again.

async get_files_total() → int[source]#: Returns a total number of files in this RemoteBox

async get_last_file_id() → int[source]#: Returns last channel file id. If nothing found returns 0

async get_requestkey(basekey: BaseKey) → RequestKey[source]#

Returns RequestKey for this RemoteBox. You should use this method if you want to decrypt other’s RemoteBox.

Arguments:

basekey (BaseKey):: To make a RequestKey for other’s RemoteBox you need to create new BaseKey for it. Later this key will be used for Box decryption.

property is_encrypted: bool#: Will return True if this is an Encrypted class, False if Decrypted

async left() → None[source]#: With calling this method you will left RemoteBox Channel.

async push_file(pf: PreparedFile, progress_callback: Callable[[int, int], None] | None = None, use_slow_upload: bool | None = False) → DecryptedRemoteBoxFile[source]#

Uploads PreparedFile to the RemoteBox.

Arguments:

pf (PreparedFile):: PreparedFile to upload. You should receive it via DecryptedLocalBox.prepare_file.
progress_callback (Callable[[int, int], None], optional):: A callback function accepting two parameters: (downloaded_bytes, total).
use_slow_upload (bool, optional):: Will use default upload function from the Telethon library instead of function from fastelethon.py. Use this if you have problems with upload.

This method used to search for files in your RemoteBox.

Arguments:

sf (SearchFilter):: SearchFilter with kwargs you like.
mainkey (MainKey, optional):: MainKey for this RemoteBox.
dlb (DecryptedLocalBox, optional):: LocalBox associated with this RemoteBox. We will take MainKey from it.
cache_preview (bytes, optional):: Will cache preview in file object if True.
return_imported_as_erbf (bool, optional):: If specified, will yield files that generator can’t decrypt (imported) as EncryptedRemoteBoxFile.
reverse (bool, optional):: If set to True, the remote files will be returned in reverse order (from newest to oldest, instead of the default oldest to newest).

Note

If dlb and mainkey not specified, then method will search only for EncryptedRemoteBoxFile.
You may ignore this kwargs if you call this method on DecryptedRemoteBox class.

async sign_files(toggle: bool) → bool[source]#

This function will enable or disable (by toggle) file(/message) sender signature.

Same as Box Channel Settings -> “Sign Messages”

Arguments:

toggle (bool):: If True, will enable signatures. If False, will disable message signing.

Will return True if file signing was enabled or False if you don’t have enough privileges to change it.

property tc: TelegramClient#: Returns TelegramClient

Updates already uploaded RemoteBox file. This will make a full reupload and Message edit.

rbf (EncryptedRemoteBoxFile, DecryptedRemoteBoxFile):: The RemoteBox file to update. We will only take a Message object from it. The rbf will NOT be updated by itself, instead, new RemoteBox file object will be returned after update.
pf (PreparedFile):: PreparedFile to upload. You should receive it via DecryptedLocalBox.prepare_file (set skip_fingerprint_check to True).
progress_callback (Callable[[int, int], None], optional):: A callback function accepting two parameters: (downloaded_bytes, total).
use_slow_upload (bool, optional):: Will use default upload function from the Telethon library instead of function from fastelethon.py. Use this if you have problems with upload.

class tgbox.api.remote.EncryptedRemoteBoxFile(id: int, erb: EncryptedRemoteBox, message_document: Message | None = None, cache_preview: bool = True, defaults_: RemoteBoxDefaults | DefaultsTableWrapper | None = None)[source]#

Bases: object

Class that represents encrypted remote file. Without decryption you can only retrieve basic information, like Prefix, Verbyte, BoxSalt, FileSalt, sender & etc.

More information you can get from docs. Typically you don’t need to use this class.

Retrieving:

from asyncio import run as asyncio_run
from tgbox.api import get_remotebox, get_localbox

async def main():
    dlb = await get_localbox(basekey)
    drb = await get_remotebox(dlb)

    erbf = await drb.get_file(
        id = await dlb.get_last_file_id(),
        decrypt = False
    )
    print(erbf.file_salt.hex())
    print(erbf.box_salt.hex())

asyncio_run(main())

property box_channel: Channel#: Returns RemoteBox Channel object.

property box_channel_id: int#: Returns ID of the RemoteBox Channel.

property box_salt: BoxSalt | None#: Returns BoxSalt or None if not initialized

async decrypt(key: MainKey | FileKey | ImportKey | None = None, drb: DecryptedRemoteBox | None = None, erase_encrypted_metadata: bool = True) → DecryptedRemoteBoxFile[source]#

Returns DecryptedRemoteBoxFile.

Arguments:

key (FileKey, MainKey, ImportKey):: Decryption key. Must be specified if drb argument is None.
drb (DecryptedRemoteBox, optional):: Decrypted RemoteBox. Must be specified if key argument is None.
erase_encrypted_metadata (bool, optional):: Will remove metadata from the parent EncryptedRemoteBoxFile after decryption to save more RAM if True. You can call .init() method on it to load it again.

property defaults: DefaultsTableWrapper | RemoteBoxDefaults#: Will return DefaultsTableWrapper or RemoteBoxDefaults.

async delete() → None[source]#: TOTALLY removes file from RemoteBox. You and all participants of the EncryptedRemoteBox will lose access to it FOREVER. This action can’t be undone. You need to have rights for this action.

Note

If you want to delete file only from your LocalBox then you can use the same delete() method on your LocalBoxFile.

property file: File#: Returns Telethon’s File object.

property file_file_name: bytes#: Returns remote file name.

property file_iv: IV | None#: Returns IV or None if not initialized

property file_salt: FileSalt | None#: Returns FileSalt or None if not initialized

property file_size: int#: Returns size of the File from Message object.

property fingerprint: bytes | None#: Returns file fingerprint (hash of file path plus mainkey) or None

async get_imported_from_entity() → Channel | User | None[source]#

Returns forward author entity (the one Document was forwarded from, i.e User / Channel).

If Author is Channel, Channel object will be returned.
If Author is User, User object will be returned.

get_requestkey(mainkey: MainKey) → RequestKey[source]#

Returns RequestKey for this file. You should use this method if you want to decrypt other’s EncryptedRemoteBoxFile.

Arguments:

mainkey (MainKey):: To make a RequestKey for other’s RemoteBoxFile you need to have your Box. Take key from your DecryptedLocalBox and specify it here.

async get_sender_entity() → Channel | User | None[source]#

Returns post author entity if “Sign Messages” with “Show author’s profiles” was enabled in Box Channel, None otherwise.

If Author is Channel, Channel object will be returned.
If Author is User, User object will be returned.

property id: int#: Returns message id.

property imported: bool#: Returns True if file was imported from other RemoteBox. False otherwise.

property imported_from_id: int | None#

Returns forward author (the entity Document was forwarded from, i.e User / Channel) ID.

If Author is Channel, ID will be negative.
If User, ID will be always positive (> 0).

async init(verify_prefix: bool = True) → EncryptedRemoteBoxFile[source]#

This method will download and set raw RemoteBoxFile metadata. If metadata length is bigger than self.defaults.METADATA_MAX then errors.LimitExceeded will be raised.

Arguments:

verify_prefix (bool, optional):: If True, will check that file has a defaults.PREFIX in metadata, and if not, will raise a NotATgboxFile exception.

property initialized: bool#: Returns True if class was initialized.

property is_encrypted: bool#: Will return True if this is an Encrypted class, False if Decrypted

property message: Message#: Returns Telethon’s Message object.

property minor_version: int | None#: Returns Minor Version of this file or None if class wasn’t initialized. If it’s a -1, then file was uploaded before the version 1.3.0 and minor is unknown.

property prefix: bytes | None#: Returns file prefix or None if not initialized

property rb: EncryptedRemoteBox | DecryptedRemoteBox#: Returns EncryptedRemoteBox from EncryptedRemoteBoxFile and DecryptedRemoteBox from DecryptedRemoteBoxFile

property sender: str | None#: Returns post author if “Sign Messages” is enabled in Box Channel, None otherwise.

property sender_id: int | None#

Returns post author ID if “Sign Messages” with “Show author’s profiles” was enabled in Box Channel, None otherwise.

If Sender is Channel, ID will be negative.
If User, ID will be always positive (> 0).

property updated_at_time: int | None#: Returns time when file was updated or None if not initialized

property upload_time: int | None#: Returns upload time or None if not initialized

property version_byte: bytes | None#: Returns Verbyte or None if not initialized

Returns EncryptedRemoteBox or DecryptedRemoteBox if you specify dlb.

Note

Must be specified at least dlb or tc with entity.

Arguments:

dlb (DecryptedLocalBox, optional):

Should be specified if tc is None.

tc (TelegramClient, optional):

Should be specified if dlb is None. entity should be specified with tc.

Note that tc must be already connected with Telegram via await tc.connect().

entity (PeerChannel, int, str, optional):

Can be Channel ID, Username or PeerChannel. Will be used if specified. Must be specified with tc.

proxy (tuple, list, dict, optional):

An iterable consisting of the proxy info. If connection is one of MTProxy, then it should contain MTProxy credentials: (‘hostname’, port, ‘secret’). Otherwise, it’s meant to store function parameters for PySocks, like (type, ‘hostname’, port). See Anorov/PySocks for more info.

Function used for making RemoteBox.

Arguments:

tc (TelegramClient):

Account to make private Telegram channel. You must be signed in via log_in().

box_name (str, optional):

Name of your Local and Remote boxes. defaults.DEF_TGBOX_NAME by default.

rb_prefix (str, optional):

Prefix of your RemoteBox. defaults.REMOTEBOX_PREFIX by default.

box_image (PathLike, optional):

PathLike to image that will be used as Channel photo of your RemoteBox.

Can be set to None if you don’t want to set Channel photo.

box_salt (BoxSalt, optional):

Random 32 bytes. Will be used in MainKey creation. Default is BoxSalt.generate().

tgbox.api.db module#

This module stores wrappers around Tgbox SQL DB.

class tgbox.api.db.SqlTableWrapper(aiosql_conn, table_name: str)[source]#

Bases: object

A low-level wrapper to SQLite Tables.

async commit() → None[source]#

async count_rows() → int[source]#: Execute SELECT count(*) from TABLE_NAME

async execute(sql_tuple: tuple, commit: bool = True)[source]#

async insert(*args, sql_statement: str | None = None, commit: bool = True) → None[source]#

If sql_statement isn’t specified, then will be used INSERT INTO TABLE_NAME values (...).

This method doesn’t check if you insert correct data or correct amount of it, you should know DB structure.

async select(sql_tuple: tuple | None = None) → AsyncGenerator[source]#: If sql_tuple isn’t specified, then will be used (SELECT * FROM TABLE_NAME, ()) statement.

async select_once(sql_tuple: tuple | None = None) → tuple[source]#: Will return first row which match the sql_tuple, see select() method for sql_tuple details.

property table_name: str#: Returns table name

class tgbox.api.db.TgboxDB(db_path: PathLike | str)[source]#

Bases: object

async close() → None[source]#: Will close TgboxDB

property closed: bool#: This method will return None if DB wasn’t opened, False if it’s still opened, True if it’s was closed.

async static create(db_path: str | PathLike) → TgboxDB[source]#: Will initialize TgboxDB

property db_path: PathLike#: Returns a path to TgboxDB file

async init() → TgboxDB[source]#

property initialized: bool#: Will return True if TgboxDB is initialized

property name: str#: Returns TgboxDB name

tgbox.api.utils module#

Module with utils for api package.

class tgbox.api.utils.DefaultsTableWrapper(tgbox_db: TgboxDB)[source]#

Bases: object

This little class will wrap around the DEFAULTS table of TGBOX DB and will fetch all contents of it.

You can await the change coroutine to change default values to your own.

async change(key: str, value) → None[source]#

This method can change the defaults values

Arguments:

key (str):: Key to change, i.e METADATA_MAX.
value:: Key’s new value.

Warning

We don’t verify here that value type corresponds to real type of Key or that value doesn’t overflow the allowed value maximum. Be sure to specify the correct Key values.

Example:

from asyncio import run as asyncio_run

from tgbox.defaults import DEF_TGBOX_NAME
from tgbox.api.db import TgboxDB
from tgbox.api.utils import DefaultsTableWrapper

async def main():
    # Make a DefaultsTableWrapper object
    tdb = await TgboxDB(DEF_TGBOX_NAME).init()
    dtw = await DefaultsTableWrapper(tdb).init()

    # Change METADATA_MAX to the max allowed size
    dtw.change('METADATA_MAX', 256**3-1)

    # Access DTW from the DecryptedLocalBox
    ... # Some code was omitted here
    # Change the default download path
    dlb.defaults.change('DOWNLOAD_PATH', 'Downloads')

asyncio_run(main())

async init() → DefaultsTableWrapper[source]#: Fetch the defaults and initialize

property initialized: bool#

class tgbox.api.utils.DirectoryRoot[source]#

Bases: object

Type used to specify that you want to access absolute local directory root.

This class doesn’t have any methods, please use it only for lbd.iterdir

class tgbox.api.utils.PreparedFile(dlb: tgbox.api.local.DecryptedLocalBox, file: BinaryIO, filekey: tgbox.keys.FileKey, filesize: int, filepath: Path, filesalt: tgbox.crypto.FileSalt, hmackey: tgbox.keys.HMACKey, fingerprint: bytes, metadata: bytes, imported: bool)[source]#

Bases: object

This dataclass store data needed for upload by DecryptedRemoteBox.push_file in future.

Usually it’s only for internal use.

dlb: tgbox.api.local.DecryptedLocalBox#

file: BinaryIO#

filekey: tgbox.keys.FileKey#

filepath: Path#

filesalt: tgbox.crypto.FileSalt#

filesize: int#

fingerprint: bytes#

hmackey: tgbox.keys.HMACKey#

imported: bool#

metadata: bytes#

set_file_id(id: int)[source]#: You should set ID after pushing to remote

set_updated_enc_metadata(ue_metadata: bytes)[source]#

If user requested to update some already pushed to Remote file AND if target file HAS Updated Encrypted Metadata in caption then we need to re-encrypt it with a new filekey and attach here.

This is for internal usage, you can ignore it.

set_upload_time(upload_time: int)[source]#: You should set time after pushing to remote

class tgbox.api.utils.RemoteBoxDefaults(METADATA_MAX: int, FILE_PATH_MAX: int, DOWNLOAD_PATH: Path, DEF_UNK_FOLDER: Path, DEF_NO_FOLDER: Path, FAST_SYNC_ENABLED: int)[source]#

Bases: object

If DefaultsTableWrapper was not passed into the *RemoteBox, this object will be used

DEF_NO_FOLDER: Path#

DEF_UNK_FOLDER: Path#

DOWNLOAD_PATH: Path#

FAST_SYNC_ENABLED: int#

FILE_PATH_MAX: int#

METADATA_MAX: int#

class tgbox.api.utils.TelegramClient(api_id: int, api_hash: str, phone_number: str | None = None, session: str | StringSession | None = None, **kwargs)[source]#

Bases: TelegramClient

A little extension to the telethon.TelegramClient.

This class inherits Telethon’s TelegramClient and support all features that has telethon.TelegramClient.

Typical usage:

from asyncio import run as asyncio_run
from tgbox.api import TelegramClient, make_remotebox
from getpass import getpass # For hidden input

PHONE_NUMBER = '+10000000000' # Your phone number
API_ID = 1234567 # Your API_ID: https://my.telegram.org
API_HASH = '00000000000000000000000000000000' # Your API_HASH

async def main():
    tc = TelegramClient(
        phone_number = PHONE_NUMBER,
        api_id = API_ID,
        api_hash = API_HASH
    )
    await tc.connect()
    await tc.send_code()

    await tc.log_in(
        code = int(input('Code: ')),
        password = getpass('Pass: ')
    )
    erb = await make_remotebox(tc)

asyncio_run(main())

action(entity: hints.EntityLike, action: Union[str, types.TypeSendMessageAction], *, delay: float = 4, auto_cancel: bool = True) → Union[_ChatAction, Coroutine]#

Returns a context-manager object to represent a “chat action”.

Chat actions indicate things like “user is typing”, “user is uploading a photo”, etc.

If the action is 'cancel', you should just await the result, since it makes no sense to use a context-manager for it.

See the example below for intended usage.

Arguments

entity (entity):

The entity where the action should be showed in.

action (str | :tl:`SendMessageAction`):

The action to show. You can either pass a instance of :tl:`SendMessageAction` or better, a string used while:

'typing': typing a text message.
'contact': choosing a contact.
'game': playing a game.
'location': choosing a geo location.
'sticker': choosing a sticker.
'record-audio': recording a voice note. You may use 'record-voice' as alias.
'record-round': recording a round video.
'record-video': recording a normal video.
'audio': sending an audio file (voice note or song). You may use 'voice' and 'song' as aliases.
'round': uploading a round video.
'video': uploading a video file.
'photo': uploading a photo.
'document': uploading a document file. You may use 'file' as alias.
'cancel': cancel any pending action in this chat.

Invalid strings will raise a ValueError.

delay (int | float):

The delay, in seconds, to wait between sending actions. For example, if the delay is 5 and it takes 7 seconds to do something, three requests will be made at 0s, 5s, and 7s to cancel the action.

auto_cancel (bool):

Whether the action should be cancelled once the context manager exists or not. The default is True, since you don’t want progress to be shown when it has already completed.

Returns

Either a context-manager object or a coroutine.

Example

# Type for 2 seconds, then send a message
async with client.action(chat, 'typing'):
    await asyncio.sleep(2)
    await client.send_message(chat, 'Hello world! I type slow ^^')

# Cancel any previous action
await client.action(chat, 'cancel')

# Upload a document, showing its progress (most clients ignore this)
async with client.action(chat, 'document') as action:
    await client.send_file(chat, zip_file, progress_callback=action.progress)

add_event_handler(callback: Callable[[Any], Any], event: EventBuilder = None)#

Registers a new event handler callback.

The callback will be called when the specified event occurs.

Arguments

callback (callable):

The callable function accepting one parameter to be used.

Note that if you have used telethon.events.register in the callback, event will be ignored, and instead the events you previously registered will be used.

event (_EventBuilder | type, optional):

The event builder class or instance to be used, for instance events.NewMessage.

If left unspecified, telethon.events.raw.Raw (the :tl:`Update` objects with no further processing) will be passed instead.

Example

from telethon import TelegramClient, events
client = TelegramClient(...)

async def handler(event):
    ...

client.add_event_handler(handler, events.NewMessage)

Builds a :tl:`ReplyInlineMarkup` or :tl:`ReplyKeyboardMarkup` for the given buttons.

Does nothing if either no buttons are provided or the provided argument is already a reply markup.

You should consider using this method if you are going to reuse the markup very often. Otherwise, it is not necessary.

This method is not asynchronous (don’t use await on it).

Arguments

buttons (hints.MarkupLike):: The button, list of buttons, array of buttons or markup to convert into a markup.

Example

from telethon import Button

markup = client.build_reply_markup(Button.inline('hi'))
# later
await client.send_message(chat, 'click me', buttons=markup)

async catch_up()#

“Catches up” on the missed updates while the client was offline. You should call this method after registering the event handlers so that the updates it loads can by processed by your script.

This can also be used to forcibly fetch new updates if there are any.

Example

await client.catch_up()

async connect() → None#

Connects to Telegram.

Note

Connect means connect and nothing else, and only one low-level request is made to notify Telegram about which layer we will be using.

Before Telegram sends you updates, you need to make a high-level request, like client.get_me() <telethon.client.users.UserMethods.get_me>, as described in https://core.telegram.org/api/updates.

Example

try:
    await client.connect()
except OSError:
    print('Failed to connect')

conversation(entity: hints.EntityLike, *, timeout: float = 60, total_timeout: float = None, max_messages: int = 100, exclusive: bool = True, replies_are_responses: bool = True) → Conversation#

Creates a Conversation <telethon.tl.custom.conversation.Conversation> with the given entity.

Note

This Conversation API has certain shortcomings, such as lacking persistence, poor interaction with other event handlers, and overcomplicated usage for anything beyond the simplest case.

If you plan to interact with a bot without handlers, this works fine, but when running a bot yourself, you may instead prefer to follow the advice from https://stackoverflow.com/a/62246569/.

This is not the same as just sending a message to create a “dialog” with them, but rather a way to easily send messages and await for responses or other reactions. Refer to its documentation for more.

Arguments

entity (entity):

The entity with which a new conversation should be opened.

timeout (int | float, optional):

The default timeout (in seconds) per action to be used. You may also override this timeout on a per-method basis. By default each action can take up to 60 seconds (the value of this timeout).

total_timeout (int | float, optional):

The total timeout (in seconds) to use for the whole conversation. This takes priority over per-action timeouts. After these many seconds pass, subsequent actions will result in asyncio.TimeoutError.

max_messages (int, optional):

The maximum amount of messages this conversation will remember. After these many messages arrive in the specified chat, subsequent actions will result in ValueError.

exclusive (bool, optional):

By default, conversations are exclusive within a single chat. That means that while a conversation is open in a chat, you can’t open another one in the same chat, unless you disable this flag.

If you try opening an exclusive conversation for a chat where it’s already open, it will raise AlreadyInConversationError.

replies_are_responses (bool, optional):

Whether replies should be treated as responses or not.

If the setting is enabled, calls to conv.get_response <telethon.tl.custom.conversation.Conversation.get_response> and a subsequent call to conv.get_reply <telethon.tl.custom.conversation.Conversation.get_reply> will return different messages, otherwise they may return the same message.

Consider the following scenario with one outgoing message, 1, and two incoming messages, the second one replying:

                    Hello! <1
2> (reply to 1) Hi!
3> (reply to 1) How are you?

And the following code:

async with client.conversation(chat) as conv:
    msg1 = await conv.send_message('Hello!')
    msg2 = await conv.get_response()
    msg3 = await conv.get_reply()

With the setting enabled, msg2 will be 'Hi!' and msg3 be 'How are you?' since replies are also responses, and a response was already returned.

With the setting disabled, both msg2 and msg3 will be 'Hi!' since one is a response and also a reply.

Returns

A Conversation <telethon.tl.custom.conversation.Conversation>.

Example

# <you> denotes outgoing messages you sent
# <usr> denotes incoming response messages
with bot.conversation(chat) as conv:
    # <you> Hi!
    conv.send_message('Hi!')

    # <usr> Hello!
    hello = conv.get_response()

    # <you> Please tell me your name
    conv.send_message('Please tell me your name')

    # <usr> ?
    name = conv.get_response().raw_text

    while not any(x.isalpha() for x in name):
        # <you> Your name didn't have any letters! Try again
        conv.send_message("Your name didn't have any letters! Try again")

        # <usr> Human
        name = conv.get_response().raw_text

    # <you> Thanks Human!
    conv.send_message('Thanks {}!'.format(name))

async delete_dialog(entity: hints.EntityLike, *, revoke: bool = False)#

Deletes a dialog (leaves a chat or channel).

This method can be used as a user and as a bot. However, bots will only be able to use it to leave groups and channels (trying to delete a private conversation will do nothing).

See also Dialog.delete() <telethon.tl.custom.dialog.Dialog.delete>.

Arguments

entity (entities):

The entity of the dialog to delete. If it’s a chat or channel, you will leave it. Note that the chat itself is not deleted, only the dialog, because you left it.

revoke (bool, optional):

On private chats, you may revoke the messages from the other peer too. By default, it’s False. Set it to True to delete the history for both.

This makes no difference for bot accounts, who can only leave groups and channels.

Returns

The :tl:`Updates` object that the request produces, or nothing for private conversations.

Example

# Deleting the first dialog
dialogs = await client.get_dialogs(5)
await client.delete_dialog(dialogs[0])

# Leaving a channel by username
await client.delete_dialog('username')

async delete_messages(entity: hints.EntityLike, message_ids: Union[hints.MessageIDLike, Sequence[hints.MessageIDLike]], *, revoke: bool = True) → Sequence[types.messages.AffectedMessages]#

Deletes the given messages, optionally “for everyone”.

See also Message.delete() <telethon.tl.custom.message.Message.delete>.

Warning

This method does not validate that the message IDs belong to the chat that you passed! It’s possible for the method to delete messages from different private chats and small group chats at once, so make sure to pass the right IDs.

Arguments

entity (entity):

From who the message will be deleted. This can actually be None for normal chats, but must be present for channels and megagroups.

message_ids (list | int | Message <telethon.tl.custom.message.Message>):

The IDs (or ID) or messages to be deleted.

revoke (bool, optional):

Whether the message should be deleted for everyone or not. By default it has the opposite behaviour of official clients, and it will delete the message for everyone.

Since 24 March 2019, you can also revoke messages of any age (i.e. messages sent long in the past) the other person sent in private conversations (and of course your messages too).

Disabling this has no effect on channels or megagroups, since it will unconditionally delete the message for everyone.

Returns

A list of :tl:`AffectedMessages`, each item being the result for the delete calls of the messages in chunks of 100 each.

Example

await client.delete_messages(chat, messages)

disconnect()#

Disconnects from Telegram.

If the event loop is already running, this method returns a coroutine that you should await on your own code; otherwise the loop is ran until said coroutine completes.

Event handlers which are currently running will be cancelled before this function returns (in order to properly clean-up their tasks). In particular, this means that using disconnect in a handler will cause code after the disconnect to never run. If this is needed, consider spawning a separate task to do the remaining work.

Example

# You don't need to use this if you used "with client"
await client.disconnect()

property disconnected: Future#

Property with a Future that resolves upon disconnection.

Example

# Wait for a disconnection to occur
try:
    await client.disconnected
except OSError:
    print('Error on disconnect')

async download_file(input_location: hints.FileLike, file: hints.OutFileLike = None, *, part_size_kb: float = None, file_size: int = None, progress_callback: hints.ProgressCallback = None, dc_id: int = None, key: bytes = None, iv: bytes = None) → bytes | None#

Low-level method to download files from their input location.

Note

Generally, you should instead use download_media. This method is intended to be a bit more low-level.

Arguments

input_location (:tl:`InputFileLocation`):

The file location from which the file will be downloaded. See telethon.utils.get_input_location source for a complete list of supported types.

file (str | file, optional):

The output file path, directory, or stream-like object. If the path exists and is a file, it will be overwritten.

If the file path is None or bytes, then the result will be saved in memory and returned as bytes.

part_size_kb (int, optional):

Chunk size when downloading files. The larger, the less requests will be made (up to 512KB maximum).

file_size (int, optional):

The file size that is about to be downloaded, if known. Only used if progress_callback is specified.

progress_callback (callable, optional):

A callback function accepting two parameters: (downloaded bytes, total). Note that the total is the provided file_size.

dc_id (int, optional):

The data center the library should connect to in order to download the file. You shouldn’t worry about this.

key (‘bytes’, optional):

In case of an encrypted upload (secret chats) a key is supplied

iv (‘bytes’, optional):

In case of an encrypted upload (secret chats) an iv is supplied

Example

# Download a file and print its header
data = await client.download_file(input_file, bytes)
print(data[:16])

async download_media(message: hints.MessageLike, file: hints.FileLike = None, *, thumb: Union[int, types.TypePhotoSize] = None, progress_callback: hints.ProgressCallback = None) → str | bytes | None#

Downloads the given media from a message object.

Note that if the download is too slow, you should consider installing cryptg (through pip install cryptg) so that decrypting the received data is done in C instead of Python (much faster).

See also Message.download_media() <telethon.tl.custom.message.Message.download_media>.

Arguments

message (Message <telethon.tl.custom.message.Message> | :tl:`Media`):

The media or message containing the media that will be downloaded.

file (str | file, optional):

The output file path, directory, or stream-like object. If the path exists and is a file, it will be overwritten. If file is the type bytes, it will be downloaded in-memory and returned as a bytestring (i.e. file=bytes, without parentheses or quotes).

progress_callback (callable, optional):

A callback function accepting two parameters: (received bytes, total).

thumb (int | :tl:`PhotoSize`, optional):

Which thumbnail size from the document or photo to download, instead of downloading the document or photo itself.

If it’s specified but the file does not have a thumbnail, this method will return None.

The parameter should be an integer index between 0 and len(sizes). 0 will download the smallest thumbnail, and len(sizes) - 1 will download the largest thumbnail. You can also use negative indices, which work the same as they do in Python’s list.

You can also pass the :tl:`PhotoSize` instance to use. Alternatively, the thumb size type str may be used.

In short, use thumb=0 if you want the smallest thumbnail and thumb=-1 if you want the largest thumbnail.

Note

The largest thumbnail may be a video instead of a photo, as they are available since layer 116 and are bigger than any of the photos.

Returns

None if no media was provided, or if it was Empty. On success the file path is returned since it may differ from the one given.

Example

path = await client.download_media(message)
await client.download_media(message, filename)
# or
path = await message.download_media()
await message.download_media(filename)

# Downloading to memory
blob = await client.download_media(message, bytes)

# Printing download progress
def callback(current, total):
    print('Downloaded', current, 'out of', total,
          'bytes: {:.2%}'.format(current / total))

await client.download_media(message, progress_callback=callback)

async download_profile_photo(entity: hints.EntityLike, file: hints.FileLike = None, *, download_big: bool = True) → str | None#

Downloads the profile photo from the given user, chat or channel.

Arguments

entity (entity):: From who the photo will be downloaded.

Note

This method expects the full entity (which has the data to download the photo), not an input variant.

It’s possible that sometimes you can’t fetch the entity from its input (since you can get errors like ChannelPrivateError) but you already have it through another call, like getting a forwarded message from it.
file (str | file, optional):: The output file path, directory, or stream-like object. If the path exists and is a file, it will be overwritten. If file is the type bytes, it will be downloaded in-memory and returned as a bytestring (i.e. file=bytes, without parentheses or quotes).
download_big (bool, optional):: Whether to use the big version of the available photos.

Returns

None if no photo was provided, or if it was Empty. On success the file path is returned since it may differ from the one given.

Example

# Download your own profile photo
path = await client.download_profile_photo('me')
print(path)

async edit_2fa(current_password: str = None, new_password: str = None, *, hint: str = '', email: str = None, email_code_callback: Callable[[int], str] = None) → bool#

Changes the 2FA settings of the logged in user.

Review carefully the parameter explanations before using this method.

Note that this method may be incredibly slow depending on the prime numbers that must be used during the process to make sure that everything is safe.

Has no effect if both current and new password are omitted.

Arguments

current_password (str, optional):

The current password, to authorize changing to new_password. Must be set if changing existing 2FA settings. Must not be set if 2FA is currently disabled. Passing this by itself will remove 2FA (if correct).

new_password (str, optional):

The password to set as 2FA. If 2FA was already enabled, current_password must be set. Leaving this blank or None will remove the password.

hint (str, optional):

Hint to be displayed by Telegram when it asks for 2FA. Leaving unspecified is highly discouraged. Has no effect if new_password is not set.

email (str, optional):

Recovery and verification email. If present, you must also set email_code_callback, else it raises ValueError.

email_code_callback (callable, optional):

If an email is provided, a callback that returns the code sent to it must also be set. This callback may be asynchronous. It should return a string with the code. The length of the code will be passed to the callback as an input parameter.

If the callback returns an invalid code, it will raise CodeInvalidError.

Returns

True if successful, False otherwise.

Example

# Setting a password for your account which didn't have
await client.edit_2fa(new_password='I_<3_Telethon')

# Removing the password
await client.edit_2fa(current_password='I_<3_Telethon')

async edit_admin(entity: hints.EntityLike, user: hints.EntityLike, *, change_info: bool = None, post_messages: bool = None, edit_messages: bool = None, delete_messages: bool = None, ban_users: bool = None, invite_users: bool = None, pin_messages: bool = None, add_admins: bool = None, manage_call: bool = None, anonymous: bool = None, is_admin: bool = None, title: str = None) → Updates#

Edits admin permissions for someone in a chat.

Raises an error if a wrong combination of rights are given (e.g. you don’t have enough permissions to grant one).

Unless otherwise stated, permissions will work in channels and megagroups.

Arguments

entity (entity):

The channel, megagroup or chat where the promotion should happen.

user (entity):

The user to be promoted.

change_info (bool, optional):

Whether the user will be able to change info.

post_messages (bool, optional):

Whether the user will be able to post in the channel. This will only work in broadcast channels.

edit_messages (bool, optional):

Whether the user will be able to edit messages in the channel. This will only work in broadcast channels.

delete_messages (bool, optional):

Whether the user will be able to delete messages.

ban_users (bool, optional):

Whether the user will be able to ban users.

invite_users (bool, optional):

Whether the user will be able to invite users. Needs some testing.

pin_messages (bool, optional):

Whether the user will be able to pin messages.

add_admins (bool, optional):

Whether the user will be able to add admins.

manage_call (bool, optional):

Whether the user will be able to manage group calls.

anonymous (bool, optional):

Whether the user will remain anonymous when sending messages. The sender of the anonymous messages becomes the group itself.

Note

Users may be able to identify the anonymous admin by its custom title, so additional care is needed when using both anonymous and custom titles. For example, if multiple anonymous admins share the same title, users won’t be able to distinguish them.

is_admin (bool, optional):

Whether the user will be an admin in the chat. This will only work in small group chats. Whether the user will be an admin in the chat. This is the only permission available in small group chats, and when used in megagroups, all non-explicitly set permissions will have this value.

Essentially, only passing is_admin=True will grant all permissions, but you can still disable those you need.

title (str, optional):

The custom title (also known as “rank”) to show for this admin. This text will be shown instead of the “admin” badge. This will only work in channels and megagroups.

When left unspecified or empty, the default localized “admin” badge will be shown.

Returns

The resulting :tl:`Updates` object.

Example

# Allowing `user` to pin messages in `chat`
await client.edit_admin(chat, user, pin_messages=True)

# Granting all permissions except for `add_admins`
await client.edit_admin(chat, user, is_admin=True, add_admins=False)

async edit_folder(entity: hints.EntitiesLike = None, folder: int | Sequence[int] = None, *, unpack=None) → Updates#

Edits the folder used by one or more dialogs to archive them.

Arguments

entity (entities):

The entity or list of entities to move to the desired archive folder.

folder (int):

The folder to which the dialog should be archived to.

If you want to “archive” a dialog, use folder=1.

If you want to “un-archive” it, use folder=0.

You may also pass a list with the same length as entities if you want to control where each entity will go.

unpack (int, optional):

If you want to unpack an archived folder, set this parameter to the folder number that you want to delete.

When you unpack a folder, all the dialogs inside are moved to the folder number 0.

You can only use this parameter if the other two are not set.

Returns

The :tl:`Updates` object that the request produces.

Example

# Archiving the first 5 dialogs
dialogs = await client.get_dialogs(5)
await client.edit_folder(dialogs, 1)

# Un-archiving the third dialog (archiving to folder 0)
await client.edit_folder(dialog[2], 0)

# Moving the first dialog to folder 0 and the second to 1
dialogs = await client.get_dialogs(2)
await client.edit_folder(dialogs, [0, 1])

# Un-archiving all dialogs
await client.edit_folder(unpack=1)

async edit_message(entity: Union[hints.EntityLike, types.Message], message: Union[int, types.Message, types.InputMessageID, str] = None, text: str = None, *, parse_mode: str = (), attributes: Sequence[types.TypeDocumentAttribute] = None, formatting_entities: List[MessageEntityUnknown | MessageEntityMention | MessageEntityHashtag | MessageEntityBotCommand | MessageEntityUrl | MessageEntityEmail | MessageEntityBold | MessageEntityItalic | MessageEntityCode | MessageEntityPre | MessageEntityTextUrl | MessageEntityMentionName | InputMessageEntityMentionName | MessageEntityPhone | MessageEntityCashtag | MessageEntityUnderline | MessageEntityStrike | MessageEntityBankCard | MessageEntitySpoiler | MessageEntityCustomEmoji | MessageEntityBlockquote] | None = None, link_preview: bool = True, file: hints.FileLike = None, thumb: hints.FileLike = None, force_document: bool = False, buttons: hints.MarkupLike | None = None, supports_streaming: bool = False, schedule: hints.DateLike = None) → types.Message#

Edits the given message to change its text or media.

See also Message.edit() <telethon.tl.custom.message.Message.edit>.

Arguments

entity (entity | Message <telethon.tl.custom.message.Message>):

From which chat to edit the message. This can also be the message to be edited, and the entity will be inferred from it, so the next parameter will be assumed to be the message text.

You may also pass a :tl:`InputBotInlineMessageID` or :tl:`InputBotInlineMessageID64`, which is the only way to edit messages that were sent after the user selects an inline query result.

message (int | Message <telethon.tl.custom.message.Message> | :tl:`InputMessageID` | str):

The ID of the message (or Message <telethon.tl.custom.message.Message> itself) to be edited. If the entity was a Message <telethon.tl.custom.message.Message>, then this message will be treated as the new text.

text (str, optional):

The new text of the message. Does nothing if the entity was a Message <telethon.tl.custom.message.Message>.

parse_mode (object, optional):

See the TelegramClient.parse_mode <telethon.client.messageparse.MessageParseMethods.parse_mode> property for allowed values. Markdown parsing will be used by default.

attributes (list, optional):

Optional attributes that override the inferred ones, like :tl:`DocumentAttributeFilename` and so on.

formatting_entities (list, optional):

A list of message formatting entities. When provided, the parse_mode is ignored.

link_preview (bool, optional):

Should the link preview be shown?

file (str | bytes | file | media, optional):

The file object that should replace the existing media in the message.

thumb (str | bytes | file, optional):

Optional JPEG thumbnail (for documents). Telegram will ignore this parameter unless you pass a .jpg file! The file must also be small in dimensions and in disk size. Successful thumbnails were files below 20kB and 320x320px. Width/height and dimensions/size ratios may be important. For Telegram to accept a thumbnail, you must provide the dimensions of the underlying media through attributes= with :tl:`DocumentAttributesVideo` or by installing the optional hachoir dependency.

force_document (bool, optional):

Whether to send the given file as a document or not.

buttons (list, custom.Button <telethon.tl.custom.button.Button>, :tl:`KeyboardButton`):

The matrix (list of lists), row list or button to be shown after sending the message. This parameter will only work if you have signed in as a bot. You can also pass your own :tl:`ReplyMarkup` here.

supports_streaming (bool, optional):

Whether the sent video supports streaming or not. Note that Telegram only recognizes as streamable some formats like MP4, and others like AVI or MKV will not work. You should convert these to MP4 before sending if you want them to be streamable. Unsupported formats will result in VideoContentTypeError.

schedule (hints.DateLike, optional):

If set, the message won’t be edited immediately, and instead it will be scheduled to be automatically edited at a later time.

Note that this parameter will have no effect if you are trying to edit a message that was sent via inline bots.

Returns

The edited Message <telethon.tl.custom.message.Message>, unless entity was a :tl:`InputBotInlineMessageID` or :tl:`InputBotInlineMessageID64` in which case this method returns a boolean.

Raises

MessageAuthorRequiredError if you’re not the author of the message but tried editing it anyway.

MessageNotModifiedError if the contents of the message were not modified at all.

MessageIdInvalidError if the ID of the message is invalid (the ID itself may be correct, but the message with that ID cannot be edited). For example, when trying to edit messages with a reply markup (or clear markup) this error will be raised.

Example

message = await client.send_message(chat, 'hello')

await client.edit_message(chat, message, 'hello!')
# or
await client.edit_message(chat, message.id, 'hello!!')
# or
await client.edit_message(message, 'hello!!!')

async edit_permissions(entity: hints.EntityLike, user: Optional[hints.EntityLike] = None, until_date: hints.DateLike = None, *, view_messages: bool = True, send_messages: bool = True, send_media: bool = True, send_stickers: bool = True, send_gifs: bool = True, send_games: bool = True, send_inline: bool = True, embed_link_previews: bool = True, send_polls: bool = True, change_info: bool = True, invite_users: bool = True, pin_messages: bool = True) → Updates#

Edits user restrictions in a chat.

Set an argument to False to apply a restriction (i.e. remove the permission), or omit them to use the default True (i.e. don’t apply a restriction).

Raises an error if a wrong combination of rights are given (e.g. you don’t have enough permissions to revoke one).

By default, each boolean argument is True, meaning that it is true that the user has access to the default permission and may be able to make use of it.

If you set an argument to False, then a restriction is applied regardless of the default permissions.

It is important to note that True does not mean grant, only “don’t restrict”, and this is where the default permissions come in. A user may have not been revoked the pin_messages permission (it is True) but they won’t be able to use it if the default permissions don’t allow it either.

Arguments

entity (entity):

The channel or megagroup where the restriction should happen.

user (entity, optional):

If specified, the permission will be changed for the specific user. If left as None, the default chat permissions will be updated.

until_date (DateLike, optional):

When the user will be unbanned.

If the due date or duration is longer than 366 days or shorter than 30 seconds, the ban will be forever. Defaults to 0 (ban forever).

view_messages (bool, optional):

Whether the user is able to view messages or not. Forbidding someone from viewing messages equals to banning them. This will only work if user is set.

send_messages (bool, optional):

Whether the user is able to send messages or not.

send_media (bool, optional):

Whether the user is able to send media or not.

send_stickers (bool, optional):

Whether the user is able to send stickers or not.

send_gifs (bool, optional):

Whether the user is able to send animated gifs or not.

send_games (bool, optional):

Whether the user is able to send games or not.

send_inline (bool, optional):

Whether the user is able to use inline bots or not.

embed_link_previews (bool, optional):

Whether the user is able to enable the link preview in the messages they send. Note that the user will still be able to send messages with links if this permission is removed, but these links won’t display a link preview.

send_polls (bool, optional):

Whether the user is able to send polls or not.

change_info (bool, optional):

Whether the user is able to change info or not.

invite_users (bool, optional):

Whether the user is able to invite other users or not.

pin_messages (bool, optional):

Whether the user is able to pin messages or not.

Returns

The resulting :tl:`Updates` object.

Example

from datetime import timedelta

# Banning `user` from `chat` for 1 minute
await client.edit_permissions(chat, user, timedelta(minutes=1),
                              view_messages=False)

# Banning `user` from `chat` forever
await client.edit_permissions(chat, user, view_messages=False)

# Kicking someone (ban + un-ban)
await client.edit_permissions(chat, user, view_messages=False)
await client.edit_permissions(chat, user)

async end_takeout(success: bool) → bool#

Finishes the current takeout session.

Arguments

success (bool):: Whether the takeout completed successfully or not.

Returns

True if the operation was successful, False otherwise.

Example

await client.end_takeout(success=False)

property flood_sleep_threshold#

async forward_messages(entity: hints.EntityLike, messages: Union[hints.MessageIDLike, Sequence[hints.MessageIDLike]], from_peer: hints.EntityLike = None, *, background: bool = None, with_my_score: bool = None, silent: bool = None, as_album: bool = None, schedule: hints.DateLike = None, drop_author: bool = None, drop_media_captions: bool = None) → Sequence[types.Message]#

Forwards the given messages to the specified entity.

If you want to “forward” a message without the forward header (the “forwarded from” text), you should use send_message with the original message instead. This will send a copy of it.

See also Message.forward_to() <telethon.tl.custom.message.Message.forward_to>.

Arguments

entity (entity):: To which entity the message(s) will be forwarded.
messages (list | int | Message <telethon.tl.custom.message.Message>):: The message(s) to forward, or their integer IDs.
from_peer (entity):: If the given messages are integer IDs and not instances of the Message class, this must be specified in order for the forward to work. This parameter indicates the entity from which the messages should be forwarded.
silent (bool, optional):: Whether the message should notify people with sound or not. Defaults to False (send with a notification sound unless the person has the chat muted). Set it to True to alter this behaviour.
background (bool, optional):: Whether the message should be forwarded in background.
with_my_score (bool, optional):: Whether forwarded should contain your game score.
as_album (bool, optional):: This flag no longer has any effect.
schedule (hints.DateLike, optional):: If set, the message(s) won’t forward immediately, and instead they will be scheduled to be automatically sent at a later time.
drop_author (bool, optional):: Whether to forward messages without quoting the original author.
drop_media_captions (bool, optional):: Whether to strip captions from media. Setting this to True requires that drop_author also be set to True.

Returns

The list of forwarded Message <telethon.tl.custom.message.Message>, or a single one if a list wasn’t provided as input.

Note that if all messages are invalid (i.e. deleted) the call will fail with MessageIdInvalidError. If only some are invalid, the list will have None instead of those messages.

Example

# a single one
await client.forward_messages(chat, message)
# or
await client.forward_messages(chat, message_id, from_chat)
# or
await message.forward_to(chat)

# multiple
await client.forward_messages(chat, messages)
# or
await client.forward_messages(chat, message_ids, from_chat)

# Forwarding as a copy
await client.send_message(chat, message)

async get_admin_log(entity: hints.EntityLike, limit: float = None, *, max_id: int = 0, min_id: int = 0, search: str = None, admins: hints.EntitiesLike = None, join: bool = None, leave: bool = None, invite: bool = None, restrict: bool = None, unrestrict: bool = None, ban: bool = None, unban: bool = None, promote: bool = None, demote: bool = None, info: bool = None, settings: bool = None, pinned: bool = None, edit: bool = None, delete: bool = None, group_call: bool = None) → _AdminLogIter#

Same as iter_admin_log(), but returns a list instead.

Example

# Get a list of deleted message events which said "heck"
events = await client.get_admin_log(channel, search='heck', delete=True)

# Print the old message before it was deleted
print(events[0].old)

async get_dialogs(limit: float = None, *, offset_date: hints.DateLike = None, offset_id: int = 0, offset_peer: hints.EntityLike = <telethon.tl.types.InputPeerEmpty object>, ignore_pinned: bool = False, ignore_migrated: bool = False, folder: int = None, archived: bool = None) → _DialogsIter#

Same as iter_dialogs(), but returns a TotalList <telethon.helpers.TotalList> instead.

Example

# Get all open conversation, print the title of the first
dialogs = await client.get_dialogs()
first = dialogs[0]
print(first.title)

# Use the dialog somewhere else
await client.send_message(first, 'hi')

# Getting only non-archived dialogs (both equivalent)
non_archived = await client.get_dialogs(folder=0)
non_archived = await client.get_dialogs(archived=False)

# Getting only archived dialogs (both equivalent)
archived = await client.get_dialogs(folder=1)
archived = await client.get_dialogs(archived=True)

async get_drafts(entity: hints.EntitiesLike = None) → hints.TotalList#

Same as iter_drafts(), but returns a list instead.

Example

# Get drafts, print the text of the first
drafts = await client.get_drafts()
print(drafts[0].text)

# Get the draft in your chat
draft = await client.get_drafts('me')
print(drafts.text)

async get_entity(entity: hints.EntitiesLike) → hints.Entity | List[hints.Entity]#

Turns the given entity into a valid Telegram :tl:`User`, :tl:`Chat` or :tl:`Channel`. You can also pass a list or iterable of entities, and they will be efficiently fetched from the network.

Arguments

entity (str | int | :tl:`Peer` | :tl:`InputPeer`):

If a username is given, the username will be resolved making an API call every time. Resolving usernames is an expensive operation and will start hitting flood waits around 50 usernames in a short period of time.

If you want to get the entity for a cached username, you should first get_input_entity(username) <get_input_entity> which will use the cache), and then use get_entity with the result of the previous call.

Similar limits apply to invite links, and you should use their ID instead.

Using phone numbers (from people in your contact list), exact names, integer IDs or :tl:`Peer` rely on a get_input_entity first, which in turn needs the entity to be in cache, unless a :tl:`InputPeer` was passed.

Unsupported types will raise TypeError.

If the entity can’t be found, ValueError will be raised.

Returns

:tl:`User`, :tl:`Chat` or :tl:`Channel` corresponding to the input entity. A list will be returned if more than one was given.

Example

from telethon import utils

me = await client.get_entity('me')
print(utils.get_display_name(me))

chat = await client.get_input_entity('username')
async for message in client.iter_messages(chat):
    ...

# Note that you could have used the username directly, but it's
# good to use get_input_entity if you will reuse it a lot.
async for message in client.iter_messages('username'):
    ...

# Note that for this to work the phone number must be in your contacts
some_id = await client.get_peer_id('+34123456789')

async get_input_entity(peer: hints.EntityLike) → types.TypeInputPeer#

Turns the given entity into its input entity version.

Most requests use this kind of :tl:`InputPeer`, so this is the most suitable call to make for those cases. Generally you should let the library do its job and don’t worry about getting the input entity first, but if you’re going to use an entity often, consider making the call:

Arguments

entity (str | int | :tl:`Peer` | :tl:`InputPeer`):

If a username or invite link is given, the library will use the cache. This means that it’s possible to be using a username that changed or an old invite link (this only happens if an invite link for a small group chat is used after it was upgraded to a mega-group).

If the username or ID from the invite link is not found in the cache, it will be fetched. The same rules apply to phone numbers ('+34 123456789') from people in your contact list.

If an exact name is given, it must be in the cache too. This is not reliable as different people can share the same name and which entity is returned is arbitrary, and should be used only for quick tests.

If a positive integer ID is given, the entity will be searched in cached users, chats or channels, without making any call.

If a negative integer ID is given, the entity will be searched exactly as either a chat (prefixed with -) or as a channel (prefixed with -100).

If a :tl:`Peer` is given, it will be searched exactly in the cache as either a user, chat or channel.

If the given object can be turned into an input entity directly, said operation will be done.

Unsupported types will raise TypeError.

If the entity can’t be found, ValueError will be raised.

Returns

:tl:`InputPeerUser`, :tl:`InputPeerChat` or :tl:`InputPeerChannel` or :tl:`InputPeerSelf` if the parameter is 'me' or 'self'.

If you need to get the ID of yourself, you should use get_me with input_peer=True) instead.

Example

# If you're going to use "username" often in your code
# (make a lot of calls), consider getting its input entity
# once, and then using the "user" everywhere instead.
user = await client.get_input_entity('username')

# The same applies to IDs, chats or channels.
chat = await client.get_input_entity(-123456789)

async get_me(input_peer: bool = False) → Union[types.User, types.InputPeerUser]#

Gets “me”, the current :tl:`User` who is logged in.

If the user has not logged in yet, this method returns None.

Arguments

input_peer (bool, optional):: Whether to return the :tl:`InputPeerUser` version or the normal :tl:`User`. This can be useful if you just need to know the ID of yourself.

Returns

Your own :tl:`User`.

Example

me = await client.get_me()
print(me.username)

async get_messages(entity: hints.EntityLike, limit: float = None, *, offset_date: hints.DateLike = None, offset_id: int = 0, max_id: int = 0, min_id: int = 0, add_offset: int = 0, search: str = None, filter: Union[types.TypeMessagesFilter, Type[types.TypeMessagesFilter]] = None, from_user: hints.EntityLike = None, wait_time: float = None, ids: Union[int, Sequence[int]] = None, reverse: bool = False, reply_to: int = None, scheduled: bool = False) → Union[_MessagesIter, _IDsIter]#

Same as iter_messages(), but returns a TotalList <telethon.helpers.TotalList> instead.

If the limit is not set, it will be 1 by default unless both min_id and max_id are set (as named arguments), in which case the entire range will be returned.

This is so because any integer limit would be rather arbitrary and it’s common to only want to fetch one message, but if a range is specified it makes sense that it should return the entirety of it.

If ids is present in the named arguments and is not a list, a single Message <telethon.tl.custom.message.Message> will be returned for convenience instead of a list.

Example

# Get 0 photos and print the total to show how many photos there are
from telethon.tl.types import InputMessagesFilterPhotos
photos = await client.get_messages(chat, 0, filter=InputMessagesFilterPhotos)
print(photos.total)

# Get all the photos
photos = await client.get_messages(chat, None, filter=InputMessagesFilterPhotos)

# Get messages by ID:
message_1337 = await client.get_messages(chat, ids=1337)

async get_participants(entity: hints.EntityLike, limit: float = None, *, search: str = '', filter: types.TypeChannelParticipantsFilter = None, aggressive: bool = False) → _ParticipantsIter#

Same as iter_participants(), but returns a TotalList <telethon.helpers.TotalList> instead.

Example

users = await client.get_participants(chat)
print(users[0].first_name)

for user in users:
    if user.username is not None:
        print(user.username)

async get_peer_id(peer: hints.EntityLike, add_mark: bool = True) → int#

Gets the ID for the given entity.

This method needs to be async because peer supports usernames, invite-links, phone numbers (from people in your contact list), etc.

If add_mark is False, then a positive ID will be returned instead. By default, bot-API style IDs (signed) are returned.

Example

print(await client.get_peer_id('me'))

async get_permissions(entity: hints.EntityLike, user: hints.EntityLike = None) → Optional[custom.ParticipantPermissions]#

Fetches the permissions of a user in a specific chat or channel or get Default Restricted Rights of Chat or Channel.

Note

This request has to fetch the entire chat for small group chats, which can get somewhat expensive, so use of a cache is advised.

Arguments

entity (entity):: The channel or chat the user is participant of.
user (entity, optional):: Target user.

Returns

A ParticipantPermissions <telethon.tl.custom.participantpermissions.ParticipantPermissions> instance. Refer to its documentation to see what properties are available.

Example

permissions = await client.get_permissions(chat, user)
if permissions.is_admin:
    # do something

# Get Banned Permissions of Chat
await client.get_permissions(chat)

async get_profile_photos(entity: hints.EntityLike, limit: int = None, *, offset: int = 0, max_id: int = 0) → _ProfilePhotoIter#

Same as iter_profile_photos(), but returns a TotalList <telethon.helpers.TotalList> instead.

Example

# Get the photos of a channel
photos = await client.get_profile_photos(channel)

# Download the oldest photo
await client.download_media(photos[-1])

async get_stats(entity: hints.EntityLike, message: Union[int, types.Message] = None)#

Retrieves statistics from the given megagroup or broadcast channel.

Note that some restrictions apply before being able to fetch statistics, in particular the channel must have enough members (for megagroups, this requires at least 500 members).

Arguments

entity (entity):: The channel from which to get statistics.
message (int | Message, optional):: The message ID from which to get statistics, if your goal is to obtain the statistics of a single message.

Raises

If the given entity is not a channel (broadcast or megagroup), a TypeError is raised.

If there are not enough members (poorly named) errors such as telethon.errors.ChatAdminRequiredError will appear.

Returns

If both entity and message were provided, returns :tl:`MessageStats`. Otherwise, either :tl:`BroadcastStats` or :tl:`MegagroupStats`, depending on whether the input belonged to a broadcast channel or megagroup.

Example

# Some megagroup or channel username or ID to fetch
channel = -100123
stats = await client.get_stats(channel)
print('Stats from', stats.period.min_date, 'to', stats.period.max_date, ':')
print(stats.stringify())

async inline_query(bot: hints.EntityLike, query: str, *, entity: hints.EntityLike = None, offset: str = None, geo_point: types.GeoPoint = None) → InlineResults#

Makes an inline query to the specified bot (@vote New Poll).

Arguments

bot (entity):

The bot entity to which the inline query should be made.

query (str):

The query that should be made to the bot.

entity (entity, optional):

The entity where the inline query is being made from. Certain bots use this to display different results depending on where it’s used, such as private chats, groups or channels.

If specified, it will also be the default entity where the message will be sent after clicked. Otherwise, the “empty peer” will be used, which some bots may not handle correctly.

offset (str, optional):

The string offset to use for the bot.

geo_point (:tl:`GeoPoint`, optional)

The geo point location information to send to the bot for localised results. Available under some bots.

Returns

A list of custom.InlineResult <telethon.tl.custom.inlineresult.InlineResult>.

Example

# Make an inline query to @like
results = await client.inline_query('like', 'Do you like Telethon?')

# Send the first result to some chat
message = await results[0].click('TelethonOffTopic')

async is_bot() → bool#

Return True if the signed-in user is a bot, False otherwise.

Example

if await client.is_bot():
    print('Beep')
else:
    print('Hello')

is_connected() → bool#

Returns True if the user has connected.

This method is not asynchronous (don’t use await on it).

Example

while client.is_connected():
    await asyncio.sleep(1)

async is_user_authorized() → bool#

Returns True if the user is authorized (logged in).

Example

if not await client.is_user_authorized():
    await client.send_code_request(phone)
    code = input('enter code: ')
    await client.sign_in(phone, code)

iter_admin_log(entity: hints.EntityLike, limit: float = None, *, max_id: int = 0, min_id: int = 0, search: str = None, admins: hints.EntitiesLike = None, join: bool = None, leave: bool = None, invite: bool = None, restrict: bool = None, unrestrict: bool = None, ban: bool = None, unban: bool = None, promote: bool = None, demote: bool = None, info: bool = None, settings: bool = None, pinned: bool = None, edit: bool = None, delete: bool = None, group_call: bool = None) → _AdminLogIter#

Iterator over the admin log for the specified channel.

The default order is from the most recent event to to the oldest.

Note that you must be an administrator of it to use this method.

If none of the filters are present (i.e. they all are None), all event types will be returned. If at least one of them is True, only those that are true will be returned.

Arguments

entity (entity):

The channel entity from which to get its admin log.

limit (int | None, optional):

Number of events to be retrieved.

The limit may also be None, which would eventually return the whole history.

max_id (int):

All the events with a higher (newer) ID or equal to this will be excluded.

min_id (int):

All the events with a lower (older) ID or equal to this will be excluded.

search (str):

The string to be used as a search query.

admins (entity | list):

If present, the events will be filtered by these admins (or single admin) and only those caused by them will be returned.

join (bool):

If True, events for when a user joined will be returned.

leave (bool):

If True, events for when a user leaves will be returned.

invite (bool):

If True, events for when a user joins through an invite link will be returned.

restrict (bool):

If True, events with partial restrictions will be returned. This is what the API calls “ban”.

unrestrict (bool):

If True, events removing restrictions will be returned. This is what the API calls “unban”.

ban (bool):

If True, events applying or removing all restrictions will be returned. This is what the API calls “kick” (restricting all permissions removed is a ban, which kicks the user).

unban (bool):

If True, events removing all restrictions will be returned. This is what the API calls “unkick”.

promote (bool):

If True, events with admin promotions will be returned.

demote (bool):

If True, events with admin demotions will be returned.

info (bool):

If True, events changing the group info will be returned.

settings (bool):

If True, events changing the group settings will be returned.

pinned (bool):

If True, events of new pinned messages will be returned.

edit (bool):

If True, events of message edits will be returned.

delete (bool):

If True, events of message deletions will be returned.

group_call (bool):

If True, events related to group calls will be returned.

Yields

Instances of AdminLogEvent <telethon.tl.custom.adminlogevent.AdminLogEvent>.

Example

async for event in client.iter_admin_log(channel):
    if event.changed_title:
        print('The title changed from', event.old, 'to', event.new)

iter_dialogs(limit: float = None, *, offset_date: hints.DateLike = None, offset_id: int = 0, offset_peer: hints.EntityLike = <telethon.tl.types.InputPeerEmpty object>, ignore_pinned: bool = False, ignore_migrated: bool = False, folder: int = None, archived: bool = None) → _DialogsIter#

Iterator over the dialogs (open conversations/subscribed channels).

The order is the same as the one seen in official applications (first pinned, them from those with the most recent message to those with the oldest message).

Arguments

limit (int | None):

How many dialogs to be retrieved as maximum. Can be set to None to retrieve all dialogs. Note that this may take whole minutes if you have hundreds of dialogs, as Telegram will tell the library to slow down through a FloodWaitError.

offset_date (datetime, optional):

The offset date to be used.

offset_id (int, optional):

The message ID to be used as an offset.

offset_peer (:tl:`InputPeer`, optional):

The peer to be used as an offset.

ignore_pinned (bool, optional):

Whether pinned dialogs should be ignored or not. When set to True, these won’t be yielded at all.

ignore_migrated (bool, optional):

Whether :tl:`Chat` that have migrated_to a :tl:`Channel` should be included or not. By default all the chats in your dialogs are returned, but setting this to True will ignore (i.e. skip) them in the same way official applications do.

folder (int, optional):

The folder from which the dialogs should be retrieved.

If left unspecified, all dialogs (including those from folders) will be returned.

If set to 0, all dialogs that don’t belong to any folder will be returned.

If set to a folder number like 1, only those from said folder will be returned.

By default Telegram assigns the folder ID 1 to archived chats, so you should use that if you need to fetch the archived dialogs.

archived (bool, optional):

Alias for folder. If unspecified, all will be returned, False implies folder=0 and True implies folder=1.

Yields

Instances of Dialog <telethon.tl.custom.dialog.Dialog>.

Example

# Print all dialog IDs and the title, nicely formatted
async for dialog in client.iter_dialogs():
    print('{:>14}: {}'.format(dialog.id, dialog.title))

iter_download(file: hints.FileLike, *, offset: int = 0, stride: int = None, limit: int = None, chunk_size: int = None, request_size: int = 524288, file_size: int = None, dc_id: int = None)#

Iterates over a file download, yielding chunks of the file.

This method can be used to stream files in a more convenient way, since it offers more control (pausing, resuming, etc.)

Note

Using a value for offset or stride which is not a multiple of the minimum allowed request_size, or if chunk_size is different from request_size, the library will need to do a bit more work to fetch the data in the way you intend it to.

You normally shouldn’t worry about this.

Arguments

file (hints.FileLike):

The file of which contents you want to iterate over.

offset (int, optional):

The offset in bytes into the file from where the download should start. For example, if a file is 1024KB long and you just want the last 512KB, you would use offset=512 * 1024.

stride (int, optional):

The stride of each chunk (how much the offset should advance between reading each chunk). This parameter should only be used for more advanced use cases.

It must be bigger than or equal to the chunk_size.

limit (int, optional):

The limit for how many chunks will be yielded at most.

chunk_size (int, optional):

The maximum size of the chunks that will be yielded. Note that the last chunk may be less than this value. By default, it equals to request_size.

request_size (int, optional):

Note that values outside the valid range will be clamped, and the final value will also be a multiple of the minimum allowed size.

file_size (int, optional):

If the file size is known beforehand, you should set this parameter to said value. Depending on the type of the input file passed, this may be set automatically.

dc_id (int, optional):

The data center the library should connect to in order to download the file. You shouldn’t worry about this.

Yields

bytes objects representing the chunks of the file if the right conditions are met, or memoryview objects instead.

Example

# Streaming `media` to an output file
# After the iteration ends, the sender is cleaned up
with open('photo.jpg', 'wb') as fd:
    async for chunk in client.iter_download(media):
        fd.write(chunk)

# Fetching only the header of a file (32 bytes)
# You should manually close the iterator in this case.
#
# "stream" is a common name for asynchronous generators,
# and iter_download will yield `bytes` (chunks of the file).
stream = client.iter_download(media, request_size=32)
header = await stream.__anext__()  # "manual" version of `async for`
await stream.close()
assert len(header) == 32

iter_drafts(entity: hints.EntitiesLike = None) → _DraftsIter#

Iterator over draft messages.

The order is unspecified.

Arguments

entity (hints.EntitiesLike, optional):: The entity or entities for which to fetch the draft messages. If left unspecified, all draft messages will be returned.

Yields

Instances of Draft <telethon.tl.custom.draft.Draft>.

Example

# Clear all drafts
async for draft in client.get_drafts():
    await draft.delete()

# Getting the drafts with 'bot1' and 'bot2'
async for draft in client.iter_drafts(['bot1', 'bot2']):
    print(draft.text)

iter_messages(entity: hints.EntityLike, limit: float = None, *, offset_date: hints.DateLike = None, offset_id: int = 0, max_id: int = 0, min_id: int = 0, add_offset: int = 0, search: str = None, filter: Union[types.TypeMessagesFilter, Type[types.TypeMessagesFilter]] = None, from_user: hints.EntityLike = None, wait_time: float = None, ids: Union[int, Sequence[int]] = None, reverse: bool = False, reply_to: int = None, scheduled: bool = False) → Union[_MessagesIter, _IDsIter]#

Iterator over the messages for the given chat.

The default order is from newest to oldest, but this behaviour can be changed with the reverse parameter.

If either search, filter or from_user are provided, :tl:`messages.Search` will be used instead of :tl:`messages.getHistory`.

Note

Telegram’s flood wait limit for :tl:`GetHistoryRequest` seems to be around 30 seconds per 10 requests, therefore a sleep of 1 second is the default for this limit (or above).

Arguments

entity (entity):

The entity from whom to retrieve the message history.

It may be None to perform a global search, or to get messages by their ID from no particular chat. Note that some of the offsets will not work if this is the case.

Note that if you want to perform a global search, you must set a non-empty search string, a filter. or from_user.

limit (int | None, optional):

Number of messages to be retrieved. Due to limitations with the API retrieving more than 3000 messages will take longer than half a minute (or even more based on previous calls).

The limit may also be None, which would eventually return the whole history.

offset_date (datetime):

Offset date (messages previous to this date will be retrieved). Exclusive.

offset_id (int):

Offset message ID (only messages previous to the given ID will be retrieved). Exclusive.

max_id (int):

All the messages with a higher (newer) ID or equal to this will be excluded.

min_id (int):

All the messages with a lower (older) ID or equal to this will be excluded.

add_offset (int):

Additional message offset (all of the specified offsets + this offset = older messages).

search (str):

The string to be used as a search query.

filter (:tl:`MessagesFilter` | type):

The filter to use when returning messages. For instance, :tl:`InputMessagesFilterPhotos` would yield only messages containing photos.

from_user (entity):

Only messages from this entity will be returned.

wait_time (int):

Wait time (in seconds) between different :tl:`GetHistoryRequest`. Use this parameter to avoid hitting the FloodWaitError as needed. If left to None, it will default to 1 second only if the limit is higher than 3000.

If the ids parameter is used, this time will default to 10 seconds only if the amount of IDs is higher than 300.

ids (int, list):

A single integer ID (or several IDs) for the message that should be returned. This parameter takes precedence over the rest (which will be ignored if this is set). This can for instance be used to get the message with ID 123 from a channel. Note that if the message doesn’t exist, None will appear in its place, so that zipping the list of IDs with the messages can match one-to-one.

Note

At the time of writing, Telegram will not return :tl:`MessageEmpty` for :tl:`InputMessageReplyTo` IDs that failed (i.e. the message is not replying to any, or is replying to a deleted message). This means that it is not possible to match messages one-by-one, so be careful if you use non-integers in this parameter.

reverse (bool, optional):

If set to True, the messages will be returned in reverse order (from oldest to newest, instead of the default newest to oldest). This also means that the meaning of offset_id and offset_date parameters is reversed, although they will still be exclusive. min_id becomes equivalent to offset_id instead of being max_id as well since messages are returned in ascending order.

You cannot use this if both entity and ids are None.

reply_to (int, optional):

If set to a message ID, the messages that reply to this ID will be returned. This feature is also known as comments in posts of broadcast channels, or viewing threads in groups.

This feature can only be used in broadcast channels and their linked megagroups. Using it in a chat or private conversation will result in telethon.errors.PeerIdInvalidError to occur.

When using this parameter, the filter and search parameters have no effect, since Telegram’s API doesn’t support searching messages in replies.

Note

This feature is used to get replies to a message in the discussion group. If the same broadcast channel sends a message and replies to it itself, that reply will not be included in the results.

scheduled (bool, optional):

If set to True, messages which are scheduled will be returned. All other parameter will be ignored for this, except entity.

Yields

Instances of Message <telethon.tl.custom.message.Message>.

Example

# From most-recent to oldest
async for message in client.iter_messages(chat):
    print(message.id, message.text)

# From oldest to most-recent
async for message in client.iter_messages(chat, reverse=True):
    print(message.id, message.text)

# Filter by sender
async for message in client.iter_messages(chat, from_user='me'):
    print(message.text)

# Server-side search with fuzzy text
async for message in client.iter_messages(chat, search='hello'):
    print(message.id)

# Filter by message type:
from telethon.tl.types import InputMessagesFilterPhotos
async for message in client.iter_messages(chat, filter=InputMessagesFilterPhotos):
    print(message.photo)

# Getting comments from a post in a channel:
async for message in client.iter_messages(channel, reply_to=123):
    print(message.chat.title, message.text)

iter_participants(entity: hints.EntityLike, limit: float = None, *, search: str = '', filter: types.TypeChannelParticipantsFilter = None, aggressive: bool = False) → _ParticipantsIter#

Iterator over the participants belonging to the specified chat.

The order is unspecified.

Arguments

entity (entity):

The entity from which to retrieve the participants list.

limit (int):

Limits amount of participants fetched.

search (str, optional):

Look for participants with this string in name/username.

filter (:tl:`ChannelParticipantsFilter`, optional):

The filter to be used, if you want e.g. only admins Note that you might not have permissions for some filter. This has no effect for normal chats or users.

Note

The filter :tl:`ChannelParticipantsBanned` will return restricted users. If you want banned users you should use :tl:`ChannelParticipantsKicked` instead.

aggressive (bool, optional):

Does nothing. This is kept for backwards-compatibility.

There have been several changes to Telegram’s API that limits the amount of members that can be retrieved, and this was a hack that no longer works.

Yields

The :tl:`User` objects returned by :tl:`GetParticipantsRequest` with an additional .participant attribute which is the matched :tl:`ChannelParticipant` type for channels/megagroups or :tl:`ChatParticipants` for normal chats.

Example

# Show all user IDs in a chat
async for user in client.iter_participants(chat):
    print(user.id)

# Search by name
async for user in client.iter_participants(chat, search='name'):
    print(user.username)

# Filter by admins
from telethon.tl.types import ChannelParticipantsAdmins
async for user in client.iter_participants(chat, filter=ChannelParticipantsAdmins):
    print(user.first_name)

iter_profile_photos(entity: hints.EntityLike, limit: int = None, *, offset: int = 0, max_id: int = 0) → _ProfilePhotoIter#

Iterator over a user’s profile photos or a chat’s photos.

The order is from the most recent photo to the oldest.

Arguments

entity (entity):

The entity from which to get the profile or chat photos.

limit (int | None, optional):

Number of photos to be retrieved.

The limit may also be None, which would eventually all the photos that are still available.

offset (int):

How many photos should be skipped before returning the first one.

max_id (int):

The maximum ID allowed when fetching photos.

Yields

Instances of :tl:`Photo`.

Example

# Download all the profile photos of some user
async for photo in client.iter_profile_photos(user):
    await client.download_media(photo)

async kick_participant(entity: hints.EntityLike, user: Optional[hints.EntityLike])#

Kicks a user from a chat.

Kicking yourself ('me') will result in leaving the chat.

Note

Attempting to kick someone who was banned will remove their restrictions (and thus unbanning them), since kicking is just ban + unban.

Arguments

entity (entity):: The channel or chat where the user should be kicked from.
user (entity, optional):: The user to kick.

Returns

Returns the service Message <telethon.tl.custom.message.Message> produced about a user being kicked, if any.

Example

# Kick some user from some chat, and deleting the service message
msg = await client.kick_participant(chat, user)
await msg.delete()

# Leaving chat
await client.kick_participant(chat, 'me')

list_event_handlers() → Sequence[Tuple[Callback, EventBuilder]]#

Lists all registered event handlers.

Returns

A list of pairs consisting of (callback, event).

Example

@client.on(events.NewMessage(pattern='hello'))
async def on_greeting(event):
    '''Greets someone'''
    await event.reply('Hi')

for callback, event in client.list_event_handlers():
    print(id(callback), type(event))

async log_in(password: str | None = None, code: int | str | None = None) → None[source]#

Logs in to Telegram to an existing user account. You should only use this if you are not signed in yet.

Arguments:

password (str, optional):: Your 2FA password. You can ignore this if you don’t enabled it yet.
code (int, optional):: The code that Telegram sent you after calling TelegramClient.send_code() method.

async log_out() → bool#

Logs out Telegram and deletes the current *.session file.

The client is unusable after logging out and a new instance should be created.

Returns

True if the operation was successful.

Example

# Note: you will need to login again!
await client.log_out()

property loop: AbstractEventLoop#

Property with the asyncio event loop used by this client.

Example

# Download media in the background
task = client.loop.create_task(message.download_media())

# Do some work
...

# Join the task (wait for it to complete)
await task

on(event: EventBuilder)#

Decorator used to add_event_handler more conveniently.

Arguments

event (_EventBuilder | type):: The event builder class or instance to be used, for instance events.NewMessage.

Example

from telethon import TelegramClient, events
client = TelegramClient(...)

# Here we use client.on
@client.on(events.NewMessage)
async def handler(event):
    ...

property parse_mode#

This property is the default parse mode used when sending messages. Defaults to telethon.extensions.markdown. It will always be either None or an object with parse and unparse methods.

When setting a different value it should be one of:

Object with parse and unparse methods.
A callable to act as the parse method.
A str indicating the parse_mode. For Markdown 'md' or 'markdown' may be used. For HTML, 'htm' or 'html' may be used.

The parse method should be a function accepting a single parameter, the text to parse, and returning a tuple consisting of (parsed message str, [MessageEntity instances]).

The unparse method should be the inverse of parse such that assert text == unparse(*parse(text)).

See :tl:`MessageEntity` for allowed message entities.

Example

# Disabling default formatting
client.parse_mode = None

# Enabling HTML as the default format
client.parse_mode = 'html'

async pin_message(entity: hints.EntityLike, message: Optional[hints.MessageIDLike], *, notify: bool = False, pm_oneside: bool = False)#

Pins a message in a chat.

The default behaviour is to not notify members, unlike the official applications.

See also Message.pin() <telethon.tl.custom.message.Message.pin>.

Arguments

entity (entity):: The chat where the message should be pinned.
message (int | Message <telethon.tl.custom.message.Message>):: The message or the message ID to pin. If it’s None, all messages will be unpinned instead.
notify (bool, optional):: Whether the pin should notify people or not.
pm_oneside (bool, optional):: Whether the message should be pinned for everyone or not. By default it has the opposite behaviour of official clients, and it will pin the message for both sides, in private chats.

Example

# Send and pin a message to annoy everyone
message = await client.send_message(chat, 'Pinotifying is fun!')
await client.pin_message(chat, message, notify=True)

async qr_login(ignored_ids: List[int] = None) → QRLogin#

Initiates the QR login procedure.

Note that you must be connected before invoking this, as with any other request.

It is up to the caller to decide how to present the code to the user, whether it’s the URL, using the token bytes directly, or generating a QR code and displaying it by other means.

See the documentation for QRLogin to see how to proceed after this.

Arguments

ignored_ids (List[int]):: List of already logged-in user IDs, to prevent logging in twice with the same user.

Returns

An instance of QRLogin.

Example

def display_url_as_qr(url):
    pass  # do whatever to show url as a qr to the user

qr_login = await client.qr_login()
display_url_as_qr(qr_login.url)

# Important! You need to wait for the login to complete!
await qr_login.wait()

# If you have 2FA enabled, `wait` will raise `telethon.errors.SessionPasswordNeededError`.
# You should except that error and call `sign_in` with the password if this happens.

remove_event_handler(callback: Callable[[Any], Any], event: EventBuilder = None) → int#

Inverse operation of add_event_handler().

If no event is given, all events for this callback are removed. Returns how many callbacks were removed.

Example

@client.on(events.Raw)
@client.on(events.NewMessage)
async def handler(event):
    ...

# Removes only the "Raw" handling
# "handler" will still receive "events.NewMessage"
client.remove_event_handler(handler, events.Raw)

# "handler" will stop receiving anything
client.remove_event_handler(handler)

async resend_code(sent_code: SentCode) → SentCode[source]#

Will send you login code again. This can be used to force Telegram send you SMS or Call to dictate code.

Arguments:

sent_code (SentCode):: Result of the tc.send_code or result of the tc.resend_code method.

Example:

tc = tgbox.api.TelegramClient(...)
sent_code = await tc.send_code()
sent_code = await tc.resend_code(sent_code)

run_until_disconnected()#

Runs the event loop until the library is disconnected.

It also notifies Telegram that we want to receive updates as described in https://core.telegram.org/api/updates. If an unexpected error occurs during update handling, the client will disconnect and said error will be raised.

Manual disconnections can be made by calling disconnect() <telethon.client.telegrambaseclient.TelegramBaseClient.disconnect> or sending a KeyboardInterrupt (e.g. by pressing Ctrl+C on the console window running the script).

If a disconnection error occurs (i.e. the library fails to reconnect automatically), said error will be raised through here, so you have a chance to except it on your own code.

If the loop is already running, this method returns a coroutine that you should await on your own code.

Note

If you want to handle KeyboardInterrupt in your code, simply run the event loop in your code too in any way, such as loop.run_forever() or await client.disconnected (e.g. loop.run_until_complete(client.disconnected)).

Example

# Blocks the current task here until a disconnection occurs.
#
# You will still receive updates, since this prevents the
# script from exiting.
await client.run_until_disconnected()

async send_code(force_sms: bool | None = False) → SentCode[source]#

Sends the Telegram code needed to login to the given phone number.

Arguments:

force_sms (bool, optional):: Whether to force sending as SMS.

async send_code_request(phone: str, *, force_sms: bool = False, _retry_count: int = 0) → types.auth.SentCode#

Sends the Telegram code needed to login to the given phone number.

Arguments

phone (str | int):: The phone to which the code will be sent.
force_sms (bool, optional):: Whether to force sending as SMS. This has been deprecated. See issue #4050 for context.

Returns

An instance of :tl:`SentCode`.

Example

phone = '+34 123 123 123'
sent = await client.send_code_request(phone)
print(sent)

Sends message with the given file to the specified entity.

Note

If the hachoir3 package (hachoir module) is installed, it will be used to determine metadata from audio and video files.

If the pillow package is installed and you are sending a photo, it will be resized to fit within the maximum dimensions allowed by Telegram to avoid errors.PhotoInvalidDimensionsError. This cannot be done if you are sending :tl:`InputFile`, however.

Arguments

entity (entity):

Who will receive the file.

file (str | bytes | file | media):

The file to send, which can be one of:

A local file path to an in-disk file. The file name will be the path’s base name.
A bytes byte array with the file’s data to send (for example, by using text.encode('utf-8')). A default file name will be used.
A bytes io.IOBase stream over the file to send (for example, by using open(file, 'rb')). Its .name property will be used for the file name, or a default if it doesn’t have one.
An external URL to a file over the internet. This will send the file as “external” media, and Telegram is the one that will fetch the media and send it.
A Bot API-like file_id. You can convert previously sent media to file IDs for later reusing with telethon.utils.pack_bot_file_id.
A handle to an existing file (for example, if you sent a message with media before, you can use its message.media as a file here).
A handle to an uploaded file (from upload_file).
A :tl:`InputMedia` instance. For example, if you want to send a dice use :tl:`InputMediaDice`, or if you want to send a contact use :tl:`InputMediaContact`.

To send an album, you should provide a list in this parameter.

If a list or similar is provided, the files in it will be sent as an album in the order in which they appear, sliced in chunks of 10 if more than 10 are given.

caption (str, optional):

Optional caption for the sent media message. When sending an album, the caption may be a list of strings, which will be assigned to the files pairwise.

force_document (bool, optional):

If left to False and the file is a path that ends with the extension of an image file or a video file, it will be sent as such. Otherwise always as a document.

mime_type (str, optional):

Custom mime type to use for the file to be sent (for example, audio/mpeg, audio/x-vorbis+ogg, etc.). It can change the type of files displayed. If not set to any value, the mime type will be determined automatically based on the file’s extension.

file_size (int, optional):

The size of the file to be uploaded if it needs to be uploaded, which will be determined automatically if not specified.

If the file size can’t be determined beforehand, the entire file will be read in-memory to find out how large it is.

clear_draft (bool, optional):

Whether the existing draft should be cleared or not.

progress_callback (callable, optional):

A callback function accepting two parameters: (sent bytes, total).

reply_to (int | Message <telethon.tl.custom.message.Message>):

Same as reply_to from send_message.

attributes (list, optional):

Optional attributes that override the inferred ones, like :tl:`DocumentAttributeFilename` and so on.

thumb (str | bytes | file, optional):

Optional JPEG thumbnail (for documents). Telegram will ignore this parameter unless you pass a .jpg file!

The file must also be small in dimensions and in disk size. Successful thumbnails were files below 20kB and 320x320px. Width/height and dimensions/size ratios may be important. For Telegram to accept a thumbnail, you must provide the dimensions of the underlying media through attributes= with :tl:`DocumentAttributesVideo` or by installing the optional hachoir dependency.

allow_cache (bool, optional):

This parameter currently does nothing, but is kept for backward-compatibility (and it may get its use back in the future).

parse_mode (object, optional):

See the TelegramClient.parse_mode <telethon.client.messageparse.MessageParseMethods.parse_mode> property for allowed values. Markdown parsing will be used by default.

formatting_entities (list, optional):

Optional formatting entities for the sent media message. When sending an album, formatting_entities can be a list of lists, where each inner list contains types.TypeMessageEntity. Each inner list will be assigned to the corresponding file in a pairwise manner with the caption. If provided, the parse_mode parameter will be ignored.

voice_note (bool, optional):

If True the audio will be sent as a voice note.

video_note (bool, optional):

If True the video will be sent as a video note, also known as a round video message.

buttons (list, custom.Button <telethon.tl.custom.button.Button>, :tl:`KeyboardButton`):

The matrix (list of lists), row list or button to be shown after sending the message. This parameter will only work if you have signed in as a bot. You can also pass your own :tl:`ReplyMarkup` here.

silent (bool, optional):

Whether the message should notify people with sound or not. Defaults to False (send with a notification sound unless the person has the chat muted). Set it to True to alter this behaviour.

background (bool, optional):

Whether the message should be send in background.

supports_streaming (bool, optional):

schedule (hints.DateLike, optional):

If set, the file won’t send immediately, and instead it will be scheduled to be automatically sent at a later time.

comment_to (int | Message <telethon.tl.custom.message.Message>, optional):

Similar to reply_to, but replies in the linked group of a broadcast channel instead (effectively leaving a “comment to” the specified message).

This parameter takes precedence over reply_to. If there is no linked chat, telethon.errors.sgIdInvalidError is raised.

ttl (int. optional):

The Time-To-Live of the file (also known as “self-destruct timer” or “self-destructing media”). If set, files can only be viewed for a short period of time before they disappear from the message history automatically.

The value must be at least 1 second, and at most 60 seconds, otherwise Telegram will ignore this parameter.

Not all types of media can be used with this parameter, such as text documents, which will fail with TtlMediaInvalidError.

nosound_video (bool, optional):

Only applicable when sending a video file without an audio track. If set to True, the video will be displayed in Telegram as a video. If set to False, Telegram will attempt to display the video as an animated gif. (It may still display as a video due to other factors.) The value is ignored if set on non-video files. This is set to True for albums, as gifs cannot be sent in albums.

send_as (entity):

Unique identifier (int) or username (str) of the chat or channel to send the message as. You can use this to send the message on behalf of a chat or channel where you have appropriate permissions. Use the GetSendAs to return the list of message sender identifiers, which can be used to send messages in the chat, This setting applies to the current message and will remain effective for future messages unless explicitly changed. To set this behavior permanently for all messages, use SaveDefaultSendAs.

message_effect_id (int, optional):

Unique identifier of the message effect to be added to the message; for private chats only

Returns

The Message <telethon.tl.custom.message.Message> (or messages) containing the sent file, or messages if a list of them was passed.

Example

# Normal files like photos
await client.send_file(chat, '/my/photos/me.jpg', caption="It's me!")
# or
await client.send_message(chat, "It's me!", file='/my/photos/me.jpg')

# Voice notes or round videos
await client.send_file(chat, '/my/songs/song.mp3', voice_note=True)
await client.send_file(chat, '/my/videos/video.mp4', video_note=True)

# Custom thumbnails
await client.send_file(chat, '/my/documents/doc.txt', thumb='photo.jpg')

# Only documents
await client.send_file(chat, '/my/photos/photo.png', force_document=True)

# Albums
await client.send_file(chat, [
    '/my/photos/holiday1.jpg',
    '/my/photos/holiday2.jpg',
    '/my/drawings/portrait.png'
])

# Printing upload progress
def callback(current, total):
    print('Uploaded', current, 'out of', total,
          'bytes: {:.2%}'.format(current / total))

await client.send_file(chat, file, progress_callback=callback)

# Dices, including dart and other future emoji
from telethon.tl import types
await client.send_file(chat, types.InputMediaDice(''))
await client.send_file(chat, types.InputMediaDice('🎯'))

# Contacts
await client.send_file(chat, types.InputMediaContact(
    phone_number='+34 123 456 789',
    first_name='Example',
    last_name='',
    vcard=''
))

async send_message(entity: hints.EntityLike, message: hints.MessageLike = '', *, reply_to: Union[int, types.Message] = None, attributes: Sequence[types.TypeDocumentAttribute] = None, parse_mode: str | None = (), formatting_entities: List[MessageEntityUnknown | MessageEntityMention | MessageEntityHashtag | MessageEntityBotCommand | MessageEntityUrl | MessageEntityEmail | MessageEntityBold | MessageEntityItalic | MessageEntityCode | MessageEntityPre | MessageEntityTextUrl | MessageEntityMentionName | InputMessageEntityMentionName | MessageEntityPhone | MessageEntityCashtag | MessageEntityUnderline | MessageEntityStrike | MessageEntityBankCard | MessageEntitySpoiler | MessageEntityCustomEmoji | MessageEntityBlockquote] | None = None, link_preview: bool = True, file: Union[hints.FileLike, Sequence[hints.FileLike]] = None, thumb: hints.FileLike = None, force_document: bool = False, clear_draft: bool = False, buttons: hints.MarkupLike | None = None, silent: bool = None, background: bool = None, supports_streaming: bool = False, schedule: hints.DateLike = None, comment_to: Union[int, types.Message] = None, nosound_video: bool = None, send_as: hints.EntityLike | None = None, message_effect_id: int | None = None) → types.Message#

Sends a message to the specified user, chat or channel.

The default parse mode is the same as the official applications (a custom flavour of markdown). **bold**, `code` or __italic__ are available. In addition you can send [links](https://example.com) and [mentions](@username) (or using IDs like in the Bot API: [mention](tg://user?id=123456789)) and pre blocks with three backticks.

Sending a /start command with a parameter (like ?start=data) is also done through this method. Simply send '/start data' to the bot.

See also Message.respond() <telethon.tl.custom.message.Message.respond> and Message.reply() <telethon.tl.custom.message.Message.reply>.

Arguments

entity (entity):

To who will it be sent.

message (str | Message <telethon.tl.custom.message.Message>):

The message to be sent, or another message object to resend.

The maximum length for a message is 35,000 bytes or 4,096 characters. Longer messages will not be sliced automatically, and you should slice them manually if the text to send is longer than said length.

reply_to (int | Message <telethon.tl.custom.message.Message>, optional):

Whether to reply to a message or not. If an integer is provided, it should be the ID of the message that it should reply to.

attributes (list, optional):

Optional attributes that override the inferred ones, like :tl:`DocumentAttributeFilename` and so on.

parse_mode (object, optional):

See the TelegramClient.parse_mode <telethon.client.messageparse.MessageParseMethods.parse_mode> property for allowed values. Markdown parsing will be used by default.

formatting_entities (list, optional):

A list of message formatting entities. When provided, the parse_mode is ignored.

link_preview (bool, optional):

Should the link preview be shown?

file (file, optional):

Sends a message with a file attached (e.g. a photo, video, audio or document). The message may be empty.

thumb (str | bytes | file, optional):

force_document (bool, optional):

Whether to send the given file as a document or not.

clear_draft (bool, optional):

Whether the existing draft should be cleared or not.

buttons (list, custom.Button <telethon.tl.custom.button.Button>, :tl:`KeyboardButton`):

The matrix (list of lists), row list or button to be shown after sending the message. This parameter will only work if you have signed in as a bot. You can also pass your own :tl:`ReplyMarkup` here.

All the following limits apply together:

There can be 100 buttons at most (any more are ignored).
There can be 8 buttons per row at most (more are ignored).
The maximum callback data per button is 64 bytes.
The maximum data that can be embedded in total is just over 4KB, shared between inline callback data and text.

silent (bool, optional):

Whether the message should notify people in a broadcast channel or not. Defaults to False, which means it will notify them. Set it to True to alter this behaviour.

background (bool, optional):

Whether the message should be send in background.

supports_streaming (bool, optional):

schedule (hints.DateLike, optional):

If set, the message won’t send immediately, and instead it will be scheduled to be automatically sent at a later time.

comment_to (int | Message <telethon.tl.custom.message.Message>, optional):

Similar to reply_to, but replies in the linked group of a broadcast channel instead (effectively leaving a “comment to” the specified message).

This parameter takes precedence over reply_to. If there is no linked chat, telethon.errors.sgIdInvalidError is raised.

nosound_video (bool, optional):

send_as (entity):

message_effect_id (int, optional):

Unique identifier of the message effect to be added to the message; for private chats only

Returns

The sent custom.Message <telethon.tl.custom.message.Message>.

Example

# Markdown is the default
await client.send_message('me', 'Hello **world**!')

# Default to another parse mode
client.parse_mode = 'html'

await client.send_message('me', 'Some <b>bold</b> and <i>italic</i> text')
await client.send_message('me', 'An <a href="https://example.com">URL</a>')
# code and pre tags also work, but those break the documentation :)
await client.send_message('me', '<a href="tg://user?id=me">Mentions</a>')

# Explicit parse mode
# No parse mode by default
client.parse_mode = None

# ...but here I want markdown
await client.send_message('me', 'Hello, **world**!', parse_mode='md')

# ...and here I need HTML
await client.send_message('me', 'Hello, <i>world</i>!', parse_mode='html')

# If you logged in as a bot account, you can send buttons
from telethon import events, Button

@client.on(events.CallbackQuery)
async def callback(event):
    await event.edit('Thank you for clicking {}!'.format(event.data))

# Single inline button
await client.send_message(chat, 'A single button, with "clk1" as data',
                          buttons=Button.inline('Click me', b'clk1'))

# Matrix of inline buttons
await client.send_message(chat, 'Pick one from this grid', buttons=[
    [Button.inline('Left'), Button.inline('Right')],
    [Button.url('Check this site!', 'https://example.com')]
])

# Reply keyboard
await client.send_message(chat, 'Welcome', buttons=[
    Button.text('Thanks!', resize=True, single_use=True),
    Button.request_phone('Send phone'),
    Button.request_location('Send location')
])

# Forcing replies or clearing buttons.
await client.send_message(chat, 'Reply to me', buttons=Button.force_reply())
await client.send_message(chat, 'Bye Keyboard!', buttons=Button.clear())

# Scheduling a message to be sent after 5 minutes
from datetime import timedelta
await client.send_message(chat, 'Hi, future!', schedule=timedelta(minutes=5))

async send_read_acknowledge(entity: hints.EntityLike, message: Union[hints.MessageIDLike, Sequence[hints.MessageIDLike]] = None, *, max_id: int = None, clear_mentions: bool = False, clear_reactions: bool = False) → bool#

Marks messages as read and optionally clears mentions.

This effectively marks a message as read (or more than one) in the given conversation.

If neither message nor maximum ID are provided, all messages will be marked as read by assuming that max_id = 0.

If a message or maximum ID is provided, all the messages up to and including such ID will be marked as read (for all messages whose ID ≤ max_id).

See also Message.mark_read() <telethon.tl.custom.message.Message.mark_read>.

Arguments

entity (entity):

The chat where these messages are located.

message (list | Message <telethon.tl.custom.message.Message>):

Either a list of messages or a single message.

max_id (int):

Until which message should the read acknowledge be sent for. This has priority over the message parameter.

clear_mentions (bool):

Whether the mention badge should be cleared (so that there are no more mentions) or not for the given entity.

If no message is provided, this will be the only action taken.

clear_reactions (bool):

Whether the reactions badge should be cleared (so that there are no more reaction notifications) or not for the given entity.

If no message is provided, this will be the only action taken.

Example

# using a Message object
await client.send_read_acknowledge(chat, message)
# ...or using the int ID of a Message
await client.send_read_acknowledge(chat, message_id)
# ...or passing a list of messages to mark as read
await client.send_read_acknowledge(chat, messages)

set_phone_number(phone_number: str) → None[source]#: Use this function if you didn’t specified phone_number on init.

set_proxy(proxy: tuple | dict)#

Changes the proxy which will be used on next (re)connection.

Method has no immediate effects if the client is currently connected.

The new proxy will take it’s effect on the next reconnection attempt:

on a call await client.connect() (after complete disconnect)
on auto-reconnect attempt (e.g, after previous connection was lost)

async set_receive_updates(receive_updates)#

Change the value of receive_updates.

This is an async method, because in order for Telegram to start sending updates again, a request must be made.

set_session(session: str | StringSession) → None[source]#: Use this function if you didn’t specified session on init.

async sign_in(phone: str = None, code: str | int = None, *, password: str = None, bot_token: str = None, phone_code_hash: str = None) → Union[types.User, types.auth.SentCode]#

Logs in to Telegram to an existing user or bot account.

You should only use this if you are not authorized yet.

This method will send the code if it’s not provided.

Note

In most cases, you should simply use start() and not this method.

Arguments

phone (str | int):: The phone to send the code to if no code was provided, or to override the phone that was previously used with these requests.
code (str | int):: The code that Telegram sent. Note that if you have sent this code through the application itself it will immediately expire. If you want to send the code, obfuscate it somehow. If you’re not doing any of this you can ignore this note.
password (str):: 2FA password, should be used if a previous call raised SessionPasswordNeededError.
bot_token (str):: Used to sign in as a bot. Not all requests will be available. This should be the hash the @BotFather gave you.
phone_code_hash (str, optional):: The hash returned by send_code_request. This can be left as None to use the last hash known for the phone to be used.

Returns

The signed in user, or the information about send_code_request().

Example

phone = '+34 123 123 123'
await client.sign_in(phone)  # send code

code = input('enter code: ')
await client.sign_in(phone, code)

async sign_up(code: str | int, first_name: str, last_name: str = '', *, phone: str = None, phone_code_hash: str = None) → types.User#: This method can no longer be used, and will immediately raise a ValueError. See issue #4050 for context.

start(phone: Callable[[], str] | str=<function AuthMethods.<lambda>>, password: Callable[[], str] | str=<function AuthMethods.<lambda>>, *, bot_token: str = None, force_sms: bool = False, code_callback: Callable[[], str | int]=None, first_name: str = 'New User', last_name: str = '', max_attempts: int = 3) → TelegramClient#

Starts the client (connects and logs in if necessary).

By default, this method will be interactive (asking for user input if needed), and will handle 2FA if enabled too.

If the event loop is already running, this method returns a coroutine that you should await on your own code; otherwise the loop is ran until said coroutine completes.

Arguments

phone (str | int | callable):: The phone (or callable without arguments to get it) to which the code will be sent. If a bot-token-like string is given, it will be used as such instead. The argument may be a coroutine.
password (str, callable, optional):: The password for 2 Factor Authentication (2FA). This is only required if it is enabled in your account. The argument may be a coroutine.
bot_token (str):: Bot Token obtained by @BotFather to log in as a bot. Cannot be specified with phone (only one of either allowed).
force_sms (bool, optional):: Whether to force sending the code request as SMS. This only makes sense when signing in with a phone.
code_callback (callable, optional):: A callable that will be used to retrieve the Telegram login code. Defaults to input(). The argument may be a coroutine.
first_name (str, optional):: The first name to be used if signing up. This has no effect if the account already exists and you sign in.
last_name (str, optional):: Similar to the first name, but for the last. Optional.
max_attempts (int, optional):: How many times the code/password callback should be retried or switching between signing in and signing up.

Returns

This TelegramClient, so initialization can be chained with .start().

Example

client = TelegramClient('anon', api_id, api_hash)

# Starting as a bot account
await client.start(bot_token=bot_token)

# Starting as a user account
await client.start(phone)
# Please enter the code you received: 12345
# Please enter your password: *******
# (You are now logged in)

# Starting using a context manager (this calls start()):
with client:
    pass

takeout(finalize: bool = True, *, contacts: bool = None, users: bool = None, chats: bool = None, megagroups: bool = None, channels: bool = None, files: bool = None, max_file_size: bool = None) → TelegramClient#

Returns a telethon-client which calls methods behind a takeout session.

It does so by creating a proxy object over the current client through which making requests will use :tl:`InvokeWithTakeoutRequest` to wrap them. In other words, returns the current client modified so that requests are done as a takeout:

Some of the calls made through the takeout session will have lower flood limits. This is useful if you want to export the data from conversations or mass-download media, since the rate limits will be lower. Only some requests will be affected, and you will need to adjust the wait_time of methods like client.iter_messages <telethon.client.messages.MessageMethods.iter_messages>.

By default, all parameters are None, and you need to enable those you plan to use by setting them to either True or False.

You should except errors.TakeoutInitDelayError as e, since this exception will raise depending on the condition of the session. You can then access e.seconds to know how long you should wait for before calling the method again.

There’s also a success property available in the takeout proxy object, so from the with body you can set the boolean result that will be sent back to Telegram. But if it’s left None as by default, then the action is based on the finalize parameter. If it’s True then the takeout will be finished, and if no exception occurred during it, then True will be considered as a result. Otherwise, the takeout will not be finished and its ID will be preserved for future usage as client.session.takeout_id <telethon.sessions.abstract.Session.takeout_id>.

Arguments

finalize (bool):: Whether the takeout session should be finalized upon exit or not.
contacts (bool):: Set to True if you plan on downloading contacts.
users (bool):: Set to True if you plan on downloading information from users and their private conversations with you.
chats (bool):: Set to True if you plan on downloading information from small group chats, such as messages and media.
megagroups (bool):: Set to True if you plan on downloading information from megagroups (channels), such as messages and media.
channels (bool):: Set to True if you plan on downloading information from broadcast channels, such as messages and media.
files (bool):: Set to True if you plan on downloading media and you don’t only wish to export messages.
max_file_size (int):: The maximum file size, in bytes, that you plan to download for each message with media.

Example

from telethon import errors

try:
    async with client.takeout() as takeout:
        await client.get_messages('me')  # normal call
        await takeout.get_messages('me')  # wrapped through takeout (less limits)

        async for message in takeout.iter_messages(chat, wait_time=0):
            ...  # Do something with the message

except errors.TakeoutInitDelayError as e:
    print('Must wait', e.seconds, 'before takeout')

async unpin_message(entity: hints.EntityLike, message: Optional[hints.MessageIDLike] = None, *, notify: bool = False)#

Unpins a message in a chat.

If no message ID is specified, all pinned messages will be unpinned.

See also Message.unpin() <telethon.tl.custom.message.Message.unpin>.

Arguments

entity (entity):: The chat where the message should be pinned.
message (int | Message <telethon.tl.custom.message.Message>):: The message or the message ID to unpin. If it’s None, all messages will be unpinned instead.

Example

# Unpin all messages from a chat
await client.unpin_message(chat)

async upload_file(file: hints.FileLike, *, part_size_kb: float = None, file_size: int = None, file_name: str = None, use_cache: type = None, key: bytes = None, iv: bytes = None, progress_callback: hints.ProgressCallback = None) → types.TypeInputFile#

Uploads a file to Telegram’s servers, without sending it.

Note

Generally, you want to use send_file instead.

This method returns a handle (an instance of :tl:`InputFile` or :tl:`InputFileBig`, as required) which can be later used before it expires (they are usable during less than a day).

Uploading a file will simply return a “handle” to the file stored remotely in the Telegram servers, which can be later used on. This will not upload the file to your own chat or any chat at all.

Arguments

file (str | bytes | file):

The path of the file, byte array, or stream that will be sent. Note that if a byte array or a stream is given, a filename or its type won’t be inferred, and it will be sent as an “unnamed application/octet-stream”.

part_size_kb (int, optional):

Chunk size when uploading files. The larger, the less requests will be made (up to 512KB maximum).

file_size (int, optional):

The size of the file to be uploaded, which will be determined automatically if not specified.

If the file size can’t be determined beforehand, the entire file will be read in-memory to find out how large it is.

file_name (str, optional):

The file name which will be used on the resulting InputFile. If not specified, the name will be taken from the file and if this is not a str, it will be "unnamed".

use_cache (type, optional):

This parameter currently does nothing, but is kept for backward-compatibility (and it may get its use back in the future).

key (‘bytes’, optional):

In case of an encrypted upload (secret chats) a key is supplied

iv (‘bytes’, optional):

In case of an encrypted upload (secret chats) an iv is supplied

progress_callback (callable, optional):

A callback function accepting two parameters: (sent bytes, total).

When sending an album, the callback will receive a number between 0 and the amount of files as the “sent” parameter, and the amount of files as the “total”. Note that the first parameter will be a floating point number to indicate progress within a file (e.g. 2.5 means it has sent 50% of the third file, because it’s between 2 and 3).

Returns

:tl:`InputFileBig` if the file size is larger than 10MB, InputSizedFile <telethon.tl.custom.inputsizedfile.InputSizedFile> (subclass of :tl:`InputFile`) otherwise.

Example

# Photos as photo and document
file = await client.upload_file('photo.jpg')
await client.send_file(chat, file)                       # sends as photo
await client.send_file(chat, file, force_document=True)  # sends as document

file.name = 'not a photo.jpg'
await client.send_file(chat, file, force_document=True)  # document, new name

# As song or as voice note
file = await client.upload_file('song.ogg')
await client.send_file(chat, file)                   # sends as song
await client.send_file(chat, file, voice_note=True)  # sends as voice note

async tgbox.api.utils.search_generator(sf: SearchFilter, it_messages: AsyncGenerator | None = None, lb: tgbox.api.local.DecryptedLocalBox | None = None, cache_preview: bool = True, reverse: bool = False, fetch_count: int | None = 100, erase_encrypted_metadata: bool = True) → AsyncGenerator[source]#

Generator used to search for files in dlb and rb. It’s only for internal use and you shouldn’t use it in your own projects.

If file is exported from other RemoteBox and was imported to your LocalBox, then you can specify dlb as lb. AsyncGenerator will try to get FileKey and decrypt EncryptedRemoteBoxFile. Otherwise imported file will be ignored.

tgbox.api.abstract module#

Module with some magic high-level API functions and classes of TGBOX.

class tgbox.api.abstract.Box(dlb: DecryptedLocalBox, drb: DecryptedRemoteBox, lazy_files: bool | None = False)[source]#

Bases: DecryptedLocalBox

The abstract.Box is an object that contains the methods from both DecryptedLocalBox and DecryptedRemoteBox classes.

Where possible, we try to use the methods from the LocalBox to take off unnecessary load, however, the BoxFile objects that Box return (for example from get_file() or files() or search_file() always make requests & downloads info from your RemoteBox (Telegram Channel). If you want to get data from your LocalBox only, then you can use a Box.dlb or similarly Box.drb for the RemoteBox only features.

Also, you can set a lazy_files kwarg to True so file obtaining methods will return a “Lazy” BoxFile objects.

“Lazy” BoxFile will not load DecryptedRemoteBoxFile until the load_drbf() call, thus, can be useful for only retrieving information about files without need to use a Box.dlb DecryptedLocalBox object.

Tip

To understand more about the TGBOX Protocol you can use a help() on every class/method from the tgbox.api package and Read The Docs: tgbox.readthedocs.io/en/latest/

Usage:

import asyncio
import tgbox

async def main():
    box = await tgbox.get_box(tgbox.keys.make_basekey(b'OZZY'))
    bf = await box.get_file(await box.get_last_file_id())

    print(bf.id, bf.file_name, bf.directory)
    await bf.download() # Download Box file

    await box.done() # Close all connections

asyncio.run(main())

Smart usage of DLB & DRB:

'''
In this example on file searching we load files
from the LocalBox, thus no requests to the Telegram
servers. Only if file match our SearchFilter we
download information from RemoteBox and then
download file. On other hand, the Box class
itself has the ``search_file()`` method, but it
loads *every single file* from servers. We don't
need this, as searching will be slow and pricey.

You may encounter such situations, don't hesitate
to use the DLB or DRB directly on need :)
'''

import asyncio
import tgbox

async def main():
    box = await tgbox.get_box(tgbox.keys.make_basekey(b'OZZY'))

    sf = tgbox.tools.SearchFilter(
        scope='/home/user/Music',
        file_path='Black Rain',
        mime='audio'
    )
    async for dlbf in box.dlb.search_file(sf):
        drbf = await box.drb.get_file(dlbf.id)
        await drbf.download()

    await box.done() # Close all connections

asyncio.run(main())

Or just use a ‘lazy_files’ kwarg!:

import asyncio
import tgbox

async def main():
    box = await tgbox.get_box(
        basekey = tgbox.keys.make_basekey(b'OZZY'),
        lazy_files = True
    )
    sf = tgbox.tools.SearchFilter(
        scope='/home/user/Music',
        file_path='Black Rain',
        mime='audio'
    )
    async for bf in box.search_file(sf):
        await bf.load_drbf()
        await bf.download()

    await box.done() # Close all connections

asyncio.run(main())

property api_hash: str | None#: Returns API_HASH.

property api_id: int | None#: Returns API_ID.

property box_channel_id: bytes | int | None#: Returns encrypted channel ID from EncryptedLocalBox and decrypted from DecryptedLocalBox.

property box_cr_time: bytes | int | None#: Returns encrypted box creation time from EncryptedLocalBox and decrypted from DecryptedLocalBox.

property box_salt: BoxSalt | None#: Returns BoxSalt or None if not initialized

Recursive iterate over all files/folders in LocalBox.

Arguments:

sfpid (bytes, optional):: Will start from this PartID if specified, else will start from “root” PIDs.
ignore_files (bool, optional):: Will not return LocalBoxFile associated with the LocalBoxDirectory if False.
erase_encrypted_metadata (bool, optional):: Will remove metadata from the parent EncryptedLocalBoxFile after decryption to save more RAM if True. You can call .init() method on it to load it again.

async static decrypt() → NoReturn#

Will return DecryptedLocalBox.

property defaults: DefaultsTableWrapper | RemoteBoxDefaults#: Returns DefaultsTableWrapper or RemoteBoxDefaults.

async delete(remote: bool | None = False, *args, **kwargs)[source]#

This method WILL DELETE Box!

See help(DecryptedLocalBox.delete) & see help(DecryptedRemoteBox.delete).

If remote is True, will be called the same method on the DecryptedRemoteBox, completely deleting ALL OF YOUR FILES AND BOX INFORMATION!

Use left() if you only want to left your Box Channel, not destroy it.

async delete_files(remote: bool | None = False, *args, **kwargs)[source]#

See help(DecryptedLocalBox.delete_files) & see help(DecryptedRemoteBox.delete_files).

If remote is True, will be called the same method on the DecryptedRemoteBox, deleting files in the Local & Remote Box. Do NOT set this kwarg to True if you don’t want to completely destroy and remove selected files from Box.

rb is auto passed to delete_files() and is None if remote is False.

async done()[source]#: Await this method when you end all work with Box, so we will clean up & close connections.

Yields every local file as EncryptedLocalBoxFile if you call it on EncryptedLocalBox and DecryptedLocalBoxFile if on DecryptedLocalBox. Works via self.get_file

Alternatively, you may use LocalBox.contents method.

Arguments:

cache_preview (bool, optional):: Cache preview in class or not.
min_id (int, optional):: Will iterate from this ID.
max_id (int, optional):: Will iterate up to this ID.
ids (int, list, optional):: ID or list with IDs you want to fetch. If specified, The min_id and max_id args will be ignored
decrypt (bool, optional):: Will return EncryptedLocalBoxFile if False, and DecryptedLocalBoxFile if True. If None, will be determined by class.
reverse (bool, optional):: If set to True, the local files will be returned in reverse order (from newest to oldest, instead of the default oldest to newest).
fetch_count (int, optional):: Amount of files generator will fetch and cache from SQLite table before return. 100 by default.
erase_encrypted_metadata (bool, optional):: Will remove metadata from the parent EncryptedLocalBoxFile after decryption to save more RAM if True. You can call .init() method on it to load it again.

async get_directory(path: Path | str) → DecryptedLocalBoxDirectory | None#

This method will make DecryptedLocalBoxDirectory from your path. If such path is not presented in LocalBox, then None will be returned.

Arguments:

path (Path, str):: Absolute path from which you want to make an DecryptedLocalBoxDirectory.

async get_file(id: int, cache_preview: bool = True, erase_encrypted_metadata: bool = True, decrypt: None = None, lazy: bool | None = None) → BoxFile[source]#

This method returns BoxFile object, which class contains the methods from the both of DecryptedLocalBoxFile and DecryptedRemoteBoxFile.

Tip

You may want to get file information only. For such case use the same method on the Box.dlb.

Arguments:

id (int):: Box file ID.
cache_preview (bool, optional):: Cache preview in class or not.
erase_encrypted_metadata (bool, optional):: Will remove metadata to save more RAM if True.
decrypt (bool, optional):: Guess what? Does nothing! Inherited methods like files() expect this kwarg, but here we don’t need it at all. Ignored.
lazy (bool, optional):: Lazy files don’t load DRBF until load_drbf() is called. If None, will use self.lazy_files

async get_files_total() → int#: Returns a total number of files in this LocalBox

async get_last_file_id() → int#: Returns last file id from the FILES table. If there is no files at all, will return 0.

get_requestkey(basekey: BaseKey) → RequestKey#

Returns RequestKey for this LocalBox. You should use this method if you want to decrypt other’s RemoteBox.

Arguments:

basekey (BaseKey):: To make a RequestKey for other’s RemoteBox you need to create new BaseKey for it. Later this key will be used for Box decryption.

get_sharekey(reqkey: RequestKey | None = None) → ShareKey#

Returns ShareKey for this Box. You should use this method if you want to share your LocalBox with other people.

Arguments:

reqkey (RequestKey, optional):: Requester’s RequestKey. If isn’t specified returns ImportKey of this box without encryption, so anyone with this key can decrypt ALL files in your Boxes.

async import_file(drbf: tgbox.api.remote.DecryptedRemoteBoxFile, file_path: str | Path | None = None) → DecryptedLocalBoxFile#

Imports file to your DecryptedLocalBox

Arguments:

drbf (DecryptedRemoteBoxFile):

Remote file you want to import.

file_path (Path, optional):

File’s path. Will be used drbf._file_path if None and if drbf was decrypted with the MainKey, otherwise self.defaults.DEF_NO_FOLDER.

This method will call a set_file_path(file_path) on the specified drbf if file_path isn’t a None.

You can change drbf file_path with set_file_path method before importing file, so you don’t need to specify it here.

async static init() → NoReturn#: Will fetch and parse data from Database.

property initialized: bool#: Returns True if you called .init()

property is_encrypted: bool#: Will return True if this is an Encrypted class, False if Decrypted

async is_synced() → bool[source]#

This method will compare Last file ID of RemoteBox with Last file ID of LocalBox, if the same, – will return True.

Please note that it’s not guaranteed to be right, as changes can be made not only to the last files in Box. If you share your Box with someone else, then consider to use Box.sync() method more often.

property mainkey: MainKey#: Will return MainKey of this Box

make_files_lazy() → None[source]#: Will make files that this Box output ‘Lazy’ (Lazy files don’t load DRBF until load_drbf())

make_files_unlazy() → None[source]#: Will make files that this Box output ‘Unlazy’ (Unlazy files load DRBF without load_drbf())

Prepares your file for RemoteBox.push_file

Arguments:

file (BinaryIO, bytes, TelegramVirtualFile):

Absolute file path length must be <= self.defaults.FILE_PATH_MAX; If file has no name and file_path is not specified, – it will be NO_FOLDER/{tgbox.tools.prbg(6).hex()}.

file_size (int, optional):

Bytelength of file. You can specify it if you already know file size.

file_path (str, Path, optional):

Must be <= self.defaults.FILE_PATH_MAX.

cattrs (Dict[str, Union[bytes, None]], optional):

The file’s custom metadata attributes that will be added to the main metadata. Specified dict will be packed with the PackedAttributes.

make_preview (bool, optional):

Will try to add file preview to the metadata if True (default).

skip_fingerprint_check (bool, optional):

If True, will skip the File Fingerprint check. Change it only if you want to update already pushed to RemoteBox file.

This is a wrapper around DecryptedRemoteBox.push_file. Will automatically use DecryptedLocalBox.prepare_file. See help() on both of these methods for additional arguments.

Arguments:

file (str, BinaryIO, bytes, TelegramVirtualFile, list):

file data to add to the LocalBox. In most cases it’s just opened file. If you want to upload something else, then you need to implement class that have read & name methods.

The method needs to know size of the file, so it will try to ask system what size of file on path file.name. If it’s impossible, method will try to seek file to EOF, if file isn’t seekable, then we try to get size by len() (as __len__ dunder). If all fails, method tries to get file.read()) (with load to RAM).

Abs file path length must be <= self.defaults.FILE_PATH_MAX; If file has no name and file_path is not specified then it will be NO_FOLDER/{prbg(6).hex()}.

We will treat this argument as path to file and auto open() here if it is specified as str.

This argument accept list of file for uploading files simultaneously. DO NOT specify too many big files! Otherwise you may receive 429 – FloodWaitError.

Note

This argument will be auto passed to prepare_file()

progress_callback (Callable[[int, int], None], optional):

A callback function accepting two parameters: (downloaded_bytes, total). A push_file kwarg.

use_slow_upload (bool, optional):

Will use default upload function from the Telethon library instead of function from fastelethon.py. Use this if you have problems with upload. A push_file kwarg.

Note

The *args and **kwargs will be redirected only to the DecryptedLocalBox.prepare_file method. See help(DecryptedLocalBox.prepare_file) for kwargs.

Returns:

A single BoxFile object or a list with BoxFile objects if file was specified as list with files.

async remove_empty_directories(part_ids: List[bytes] | None = None)#

By default, the Protocol will not automatically remove empty Directories after the file deletion process was done (unless forced). To do this, use this method.

Arguments:

part_ids (List[bytes], optional):

List of Part ID you want to check. Will remove only if Directory attached to this Part ID is orphaned (no files/other Directories linked)

If not specified, will check every Part ID in LocalBox.

async replace_session(basekey: BaseKey, tc: TelegramClient) → None#

This method will replace LocalBox session to session of specified TelegramClient.

Arguments:

basekey (BaseKey):: BaseKey of this LocalBox.
tc (TelegramClient):: TelegramClient from which we will extract new session.

Use this method search for files in your DecryptedLocalBox.

Arguments:

sf (SearchFilter):: SearchFilter with kwargs you like.
cache_preview (bool, optional):: Will cache preview in file object if True.
reverse (bool, optional):: If set to True, the local files will be searched in reverse order (from newest to oldest, instead of the default oldest to newest).
fetch_count (int, optional):: Amount of files generator will fetch and cache from SQLite table before return. 100 by default.
erase_encrypted_metadata (bool, optional):: Will remove metadata from the parent EncryptedLocalBoxFile after decryption to save more RAM if True. You can call .init() method on it to load it again.

property session: bytes | str | None#: Returns encrypted session from EncryptedLocalBox and decrypted from DecryptedLocalBox.

async sync(*args, **kwargs)[source]#: See help(DecryptedLocalBox.sync). drb is auto passed to sync().

property tgbox_db: TgboxDB#: Returns TgboxDB.

Bases: DecryptedLocalBoxFile

The abstract.BoxFile is an object that contains the methods from both DecryptedLocalBoxFile and DecryptedRemoteBoxFile classes.

Where possible, we try to use the methods from the LocalBoxFile to take off unnecessary load. You can access BoxFile.dlb, BoxFile.drb, BoxFile.dlbf and BoxFile.drbf from this class if you need to use methods explicitly.

BoxFile can be “Lazy”. Such objects don’t load the DecryptedLocalBoxFile until load_drbf() call.

Note

This class must be initialized firstly via init() coro.

Tip

To understand more about the TGBOX Protocol you can use a help() on every class/method from the tgbox.api package and Read The Docs: tgbox.readthedocs.io/en/latest/

property box_salt: BoxSalt | None#: Returns box salt or None if file wasn’t initialized

property cattrs: bytes | None#: Returns file Custom Attributes

async static decrypt() → NoReturn#

Returns decrypted by key/dlb EncryptedLocalBoxFile

Arguments:

key (FileKey, MainKey, ImportKey):: Decryption key. Must be specified if dlb argument is None.
dlb (DecryptedLocalBox, optional):: Decrypted LocalBox. Must be specified if key argument is None.
erase_encrypted_metadata (bool, optional):: Will remove metadata from the parent EncryptedLocalBoxFile after decryption to save more RAM if True. You can call .init() method on it to load it again.

property defaults: DefaultsTableWrapper | RemoteBoxDefaults#: Will return DefaultsTableWrapper or RemoteBoxDefaults.

async delete(remote: bool | None = False, *args, **kwargs)[source]#

See help(DecryptedLocalBoxFile.delete) & see help(DecryptedRemoteBoxFile.delete).

If remote is True, will be called the same method on the DecryptedRemoteBoxFile, deleting file in the Local & Remote Box. Do NOT set this kwarg to True if you don’t want to completely destroy and remove from Box your uploaded file.

property directory: EncryptedLocalBoxDirectory | DecryptedLocalBoxDirectory | None#: Returns None i you call it on EncryptedLocalBoxFile that wasn’t initialized, EncryptedLocalBoxDirectory if on initialized or DecryptedLocalBoxDirectory if you call it on DecryptedLocalBoxFile

property dirkey: DirectoryKey | None#: Returns DirectoryKey of this file if present.

disable_cache_preview() → None#: Sets self._cache_preview to False and removes cached preview from memory.

property download_path: Path#: Returns current download path

property duration: int#: Returns media file duration.

enable_cache_preview() → None#: Sets self._cache_preview to True. Preview will be cached after first object.preview call.

async exists(*args, **kwargs)[source]#: See help(DecryptedRemoteBox.file_exists). id is auto passed to file_exists().

property file_iv: bytes | None#: Returns file IV or None if file wasn’t initialized

property file_name: str#: Returns file name.

property file_path: Path#: Returns file path.

property file_salt: FileSalt | None#: Returns file salt or None if file wasn’t initialized

property filekey: FileKey#: Returns FileKey of this file.

property fingerprint: bytes | None#: Returns file fingerprint (hash of file path plus mainkey) or None

get_requestkey(mainkey: MainKey) → RequestKey#

Returns RequestKey for this File. You should use this method if you want to decrypt other’s EncryptedLocalBoxFile.

Arguments:

mainkey (MainKey):: To make a RequestKey for other’s EncryptedLocalBoxFile you need to have your own Box. Take key from it and specify here.

get_sharekey(reqkey: RequestKey | None = None) → ShareKey#

Returns ShareKey for this file. You should use this method if you want to share your DecryptedLocalBoxFile with other people.

Arguments:

reqkey (RequestKey, optional):: Requester’s RequestKey. If isn’t specified returns ShareKey of this file without encryption, so ANYONE with this key can decrypt this local & remote box file.

property has_hmac_sha256: bool#: Will return True if file has HMAC (v1.5+)

property hmackey: HMACKey | None#: Returns HMACKey of this file if present.

property id: int | None#: Returns file ID or None if file wasn’t initialized

property imported: bool | None#: Returns True if file was forwarded to your BoxChannel.

async init() → BoxFile[source]#

Will initialize BoxFile object. Part of initialization is downloading information about file from the Telegram servers. If you don’t want this, – use Box.dlb.

You can’t access DLBF/DRBF methods before the initialization. Call this firstly.

property initialized: bool#: Returns True if you called .init()

property is_encrypted: bool#: Will return True if this is an Encrypted class, False if Decrypted

property lb: EncryptedLocalBox | DecryptedLocalBox#: Will return EncryptedLocalBox from the EncryptedLocalBoxFile and DecryptedLocalBox from the DecryptedLocalBoxFile object.

async load_drbf()[source]#: Will load and set self.drbf if it’s None

property mime: bytes | None#: Returns mime type of the file

property minor_version: int | None#: Returns Minor Version of this file or None if class wasn’t initialized. If it’s a -1, then file was uploaded before the version 1.3.0 and minor is unknown.

property prefix: bytes | None#: Returns file prefix or None if file wasn’t initialized

property preview: bytes | None#: Returns preview bytes or b'' if cache_preview is False.

property residual_metadata: dict#: Will return metadata that left after parsing secret_metadata. This can be useful in future, when lower version will read file of a higher version.

set_download_path(path: Path)#: Will set download path to specified.

property size: int#: Returns file size (no metadata included).

async update(*args, **kwargs) → BoxFile[source]#

See help(DecryptedRemoteBox.update_file). rbf is auto passed to update_file().

self will be NOT updated! Instead, a new BoxFile object will be returned.

async update_metadata(*args, **kwargs)[source]#: See help(DecryptedLocalBoxFile.update_metadata). drbf is auto passed to update_metadata().

property upload_time: bytes | int | None#: Returns encrypted upload_time from EncryptedLocalBoxFile and decrypted from DecryptedLocalBoxFile. None if class wasn’t initialized.

property version_byte: bytes | None#: Returns Verbyte of this file or None if class wasn’t initialized

Return Box object. See help(tgbox.api.abstract.Box)

Arguments:

basekey (BaseKey):

BaseKey of your Box

tgbox_db_path (PathLike, str, optional):

PathLike to your TgboxDB (LocalBox). Default is defaults.DEF_TGBOX_NAME.

proxy (tuple, list, dict, optional):

lazy_files (bool, optional):

If True, files returned by this Box will not load DecryptedRemoteBoxFile until the load_drbf method will be called on target BoxFile. Should be useful if you only want to fetch information about files.

You can “lazy” files via make_files_lazy() and “unlazy” via make_files_unlazy() method respectively (on Box).

Makes Box object. See help(tgbox.api.abstract.Box)

Arguments:

tc (TelegramClient):

Account to make private Telegram channel. You must be signed in via log_in().

basekey (BaseKey):

BaseKey that will be used for MainKey creation.

box_name (str, optional):

Filename of your LocalBox database. If not specified, will be used defaults.DEF_TGBOX_NAME

rb_prefix (str, optional):

Prefix of your RemoteBox. defaults.REMOTEBOX_PREFIX by default.

box_image (PathLike, optional):

PathLike to image that will be used as Channel photo of your RemoteBox.

Can be set to '' (empty string) if you don’t want to set Channel photo.

Default is defaults.BOX_IMAGE_PATH

box_path (PathLike, str, optional):

Path in which we will make a database file. Current Working Dir if not specified.

box_salt (BoxSalt, optional):

Random 32 bytes. Will be used in MainKey creation. Default is BoxSalt.generate().

lazy_files (bool, optional):

You can “lazy” files via make_files_lazy() and “unlazy” via make_files_unlazy() method respectively (on Box).

tgbox.api.sync module#

This (slightly changed module & its features) was taken from the Telethon library made by Lonami under MIT License: github.com/LonamiWebs/Telethon

Parts of this file i moved to the tgbox.api.utils package module, check the out _syncify_wrap_func() and syncify functions. They are NOT mine.

Thanks to the Lonami. See part of the original description:

—> This magical module will rewrite all public methods in the public interface of the library so they can run the loop on their own if it’s not already running. This rewrite may not be desirable if the end user always uses the methods they way they should be ran, but it’s incredibly useful for quick scripts and the runtime overhead is relatively low.<—

All you should do is to firstly import this module, then anything you want.

tgbox.api.sync.sync_agen(async_gen: AsyncGenerator)[source]#

This will make async generator to sync generator, so we can write “for” loop.

Use this functions on generators that you want to syncify. For example, if you want to iterate over LocalBox in sync code (to load only local files):

... # Some code was omitted

async_gen = box.dlb.files(reverse=True)
for dlbf in tgbox.api.sync.sync_agen(async_gen):
    print(dlbf.id, dlbf.file_name, dlbf.size)

Tip

To sync coroutines you can use a sync func from the tgbox package (tgbox.sync) or use it from here as tgbox.api.sync.sync_coro

tgbox.api.sync.sync_coro(coroutine: Coroutine)#: Will call asynchronous function in current asyncio loop and return result.

tgbox.crypto module#

This module stores all cryptography used in API.

class tgbox.crypto.AESwState(key: bytes | Key, iv: IV | bytes | None = None)[source]#

Bases: object

Wrapper around AES CBC which preserve state.

Note

You should use only encrypt() or decrypt() method per one object.

decrypt(data: bytes, unpad: bool = True) → bytes[source]#

Decrypts data with AES CBC.

data length must be evenly divisible by 16.

encrypt(data: bytes, pad: bool = True, concat_iv: bool = True) → bytes[source]#

Encrypts data with AES CBC.

If concat_iv is True, then first 16 bytes of result will be IV.

property mode: int#: Returns 1 if mode is encryption and 2 if decryption.

class tgbox.crypto.BoxSalt(salt: bytes | memoryview)[source]#

Bases: Salt

This is a class-wrapper for BoxSalt

classmethod generate(bytelength: int | None = 32)#

Generates Salt by bytelength

Arguments:

bytelength (int, optional):: Bytelength of Salt. 32 bytes by default.

hex() → str#: Returns Salt as hexadecimal

class tgbox.crypto.FileSalt(salt: bytes | memoryview)[source]#

Bases: Salt

This is a class-wrapper for FileSalt

classmethod generate(bytelength: int | None = 32)#

Generates Salt by bytelength

Arguments:

bytelength (int, optional):: Bytelength of Salt. 32 bytes by default.

hex() → str#: Returns Salt as hexadecimal

class tgbox.crypto.IV(iv: bytes | memoryview)[source]#

Bases: object

This is a class-wrapper for AES IV

classmethod generate(bytelength: int | None = 16)[source]#

Generates AES IV by bytelength

Arguments:

bytelength (int, optional):: Bytelength of IV. 16 bytes by default.

hex() → str[source]#: Returns IV as hexadecimal

class tgbox.crypto.Salt(salt: bytes | memoryview)[source]#

Bases: object

This is a class-wrapper for some TGBOX salt

classmethod generate(bytelength: int | None = 32)[source]#

Generates Salt by bytelength

Arguments:

bytelength (int, optional):: Bytelength of Salt. 32 bytes by default.

hex() → str[source]#: Returns Salt as hexadecimal

tgbox.crypto.get_rnd_bytes(length: int = 32) → bytes[source]#: Returns os.urandom(length).

tgbox.errors module#

This module stores all Tgbox-unique exceptions.

exception tgbox.errors.AESError(message=None)[source]#

Bases: TgboxException

Invalid AES configuration

add_note()#: Exception.add_note(note) – add a note to the exception

args#

with_traceback()#: Exception.with_traceback(tb) – set self.__traceback__ to tb and return self.

exception tgbox.errors.AlreadyImported(message=None)[source]#

Bases: TgboxException

LocalBox have file with same ID

add_note()#: Exception.add_note(note) – add a note to the exception

args#

with_traceback()#: Exception.with_traceback(tb) – set self.__traceback__ to tb and return self.

exception tgbox.errors.BrokenDatabase(message=None)[source]#

Bases: TgboxException

Can’t parse SQLite DB

add_note()#: Exception.add_note(note) – add a note to the exception

args#

with_traceback()#: Exception.with_traceback(tb) – set self.__traceback__ to tb and return self.

exception tgbox.errors.ConcatError(message=None)[source]#

Bases: TgboxException

You must concat metadata before using OpenPretender

add_note()#: Exception.add_note(note) – add a note to the exception

args#

with_traceback()#: Exception.with_traceback(tb) – set self.__traceback__ to tb and return self.

exception tgbox.errors.DurationImpossible(message=None)[source]#

Bases: TgboxException

Can’t get media duration

add_note()#: Exception.add_note(note) – add a note to the exception

args#

with_traceback()#: Exception.with_traceback(tb) – set self.__traceback__ to tb and return self.

exception tgbox.errors.FastSyncDisabled(message=None)[source]#

Bases: TgboxException

Fast sync was disabled in Box defaults

add_note()#: Exception.add_note(note) – add a note to the exception

args#

with_traceback()#: Exception.with_traceback(tb) – set self.__traceback__ to tb and return self.

exception tgbox.errors.FingerprintExists(message=None)[source]#

Bases: TgboxException

File with the same file path already uploaded to the Box

add_note()#: Exception.add_note(note) – add a note to the exception

args#

with_traceback()#: Exception.with_traceback(tb) – set self.__traceback__ to tb and return self.

exception tgbox.errors.InUseException(message=None)[source]#

Bases: TgboxException

The DB already exists and in use

add_note()#: Exception.add_note(note) – add a note to the exception

args#

with_traceback()#: Exception.with_traceback(tb) – set self.__traceback__ to tb and return self.

exception tgbox.errors.IncorrectKey(message=None)[source]#

Bases: TgboxException

Specified key is invalid.

add_note()#: Exception.add_note(note) – add a note to the exception

args#

with_traceback()#: Exception.with_traceback(tb) – set self.__traceback__ to tb and return self.

exception tgbox.errors.InvalidFile(message=None)[source]#

Bases: TgboxException

Specified file is invalid

add_note()#: Exception.add_note(note) – add a note to the exception

args#

with_traceback()#: Exception.with_traceback(tb) – set self.__traceback__ to tb and return self.

exception tgbox.errors.LimitExceeded(message=None)[source]#

Bases: TgboxException

Value is out of allowed range

add_note()#: Exception.add_note(note) – add a note to the exception

args#

with_traceback()#: Exception.with_traceback(tb) – set self.__traceback__ to tb and return self.

exception tgbox.errors.ModeInvalid(message=None)[source]#

Bases: TgboxException

You should use only decryption or encryption per class

add_note()#: Exception.add_note(note) – add a note to the exception

args#

with_traceback()#: Exception.with_traceback(tb) – set self.__traceback__ to tb and return self.

exception tgbox.errors.NoPlaceLeftForMetadata(message=None)[source]#

Bases: TgboxException

Your edited metadata overflow Telegram caption limit

add_note()#: Exception.add_note(note) – add a note to the exception

args#

with_traceback()#: Exception.with_traceback(tb) – set self.__traceback__ to tb and return self.

exception tgbox.errors.NotATgboxFile(message=None)[source]#

Bases: TgboxException

This Telegram message isn’t a TGBOX file

add_note()#: Exception.add_note(note) – add a note to the exception

args#

with_traceback()#: Exception.with_traceback(tb) – set self.__traceback__ to tb and return self.

exception tgbox.errors.NotEnoughRights(message=None)[source]#

Bases: TgboxException

You don’t have rights for this action

add_note()#: Exception.add_note(note) – add a note to the exception

args#

with_traceback()#: Exception.with_traceback(tb) – set self.__traceback__ to tb and return self.

exception tgbox.errors.NotImported(message=None)[source]#

Bases: TgboxException

The file you try to retrieve wasn’t imported yet

add_note()#: Exception.add_note(note) – add a note to the exception

args#

with_traceback()#: Exception.with_traceback(tb) – set self.__traceback__ to tb and return self.

exception tgbox.errors.NotInitializedError(message=None)[source]#

Bases: TgboxException

The class you try to use isn’t initialized

add_note()#: Exception.add_note(note) – add a note to the exception

args#

with_traceback()#: Exception.with_traceback(tb) – set self.__traceback__ to tb and return self.

exception tgbox.errors.PathIsDirectory(message=None)[source]#

Bases: TgboxException

Specified path is Directory

add_note()#: Exception.add_note(note) – add a note to the exception

args#

with_traceback()#: Exception.with_traceback(tb) – set self.__traceback__ to tb and return self.

exception tgbox.errors.PreviewImpossible(message=None)[source]#

Bases: TgboxException

Can’t create file preview

add_note()#: Exception.add_note(note) – add a note to the exception

args#

with_traceback()#: Exception.with_traceback(tb) – set self.__traceback__ to tb and return self.

exception tgbox.errors.RemoteBoxInaccessible(message=None)[source]#

Bases: TgboxException

The RemoteBox you try to use is inaccessible

add_note()#: Exception.add_note(note) – add a note to the exception

args#

with_traceback()#: Exception.with_traceback(tb) – set self.__traceback__ to tb and return self.

exception tgbox.errors.RemoteFileNotFound(message=None)[source]#

Bases: TgboxException

Seems that there is no requested by you file

add_note()#: Exception.add_note(note) – add a note to the exception

args#

with_traceback()#: Exception.with_traceback(tb) – set self.__traceback__ to tb and return self.

exception tgbox.errors.SessionUnregistered(message=None)[source]#

Bases: TgboxException

Session you trying to use was disconnected

add_note()#: Exception.add_note(note) – add a note to the exception

args#

with_traceback()#: Exception.with_traceback(tb) – set self.__traceback__ to tb and return self.

exception tgbox.errors.TgboxException(message=None)[source]#

Bases: Exception

Base TGBOX Exception

add_note()#: Exception.add_note(note) – add a note to the exception

args#

with_traceback()#: Exception.with_traceback(tb) – set self.__traceback__ to tb and return self.

tgbox.fastelethon module#

Copied from Mautrix-Telegram: github.com/tulir/mautrix-telegram/
Copied again from Painor GitHub: gist.github.com/painor/7e74de80ae0c819d3e9abcf9989a8dd6

This file was patched for TGBOX project (github.com/NonProjects/tgbox)

and may not work as you expect in your code.

Big thanks to all contributors of this module.

class tgbox.fastelethon.DownloadSender(client: TelegramClient, sender: MTProtoSender, file: Document | Photo | InputDocumentFileLocation | InputPeerPhotoFileLocation | InputFileLocation | InputPhotoFileLocation, offset: int, limit: int, stride: int, count: int)[source]#

Bases: object

client: TelegramClient#

disconnect() → Awaitable[None][source]#

async next() → bytes | None[source]#

remaining: int#

request: GetFileRequest#

sender: MTProtoSender#

stride: int#

class tgbox.fastelethon.ParallelTransferrer(client: TelegramClient, dc_id: int | None = None)[source]#

Bases: object

auth_key: AuthKey#

client: TelegramClient#

dc_id: int#

async download(file: Document | Photo | InputDocumentFileLocation | InputPeerPhotoFileLocation | InputFileLocation | InputPhotoFileLocation, file_size: int, part_size_kb: int | None = None, offset: int | None = None, connection_count: int | None = None) → AsyncGenerator[bytes, None][source]#

async finish_upload() → None[source]#

async init_upload(file_id: int, file_size: int, part_size_kb: float | None = None, connection_count: int | None = None) → Tuple[int, int, bool][source]#

loop: AbstractEventLoop#

senders: List[DownloadSender | UploadSender] | None#

async upload(part: bytes) → None[source]#

upload_ticker: int#

class tgbox.fastelethon.UploadSender(client: TelegramClient, sender: MTProtoSender, file_id: int, part_count: int, big: bool, index: int, stride: int, loop: AbstractEventLoop)[source]#

Bases: object

client: TelegramClient#

async disconnect() → None[source]#

loop: AbstractEventLoop#

async next(data: bytes) → None[source]#

part_count: int#

previous: Task | None#

request: SaveFilePartRequest | SaveBigFilePartRequest#

sender: MTProtoSender#

stride: int#

async tgbox.fastelethon.download_file(client: TelegramClient, location: Document | Photo | InputDocumentFileLocation | InputPeerPhotoFileLocation | InputFileLocation | InputPhotoFileLocation, request_size: int = 524288, offset: int = 0, progress_callback: callable = None) → AsyncGenerator[bytes, None][source]#

async tgbox.fastelethon.stream_file(file_to_stream: BinaryIO, chunk_size=1024)[source]#: file_to_stream.read can be coroutine.

async tgbox.fastelethon.upload_file(client: TelegramClient, file: BinaryIO, progress_callback: callable = None, file_name: str = 'document', file_size: int = None, part_size_kb: int = 512) → InputFile | InputFileBig | InputFileStoryDocument[source]#

tgbox.keys module#

This module stores all keys and key making functions.

class tgbox.keys.BaseKey(key: bytes)[source]#

Bases: Key

This Key used for MainKey creation and cloned RemoteBox decryption. In API it’s usually result of keys.make_basekey.

Decodes Key by prefix and returns Key in one of Key classes.

B: BaseKey M: MainKey R: RequestKey S: ShareKey I: ImportKey F: FileKey E: EncryptedMainkey D: DirectoryKey H: HMACKey

Key example:: MSGVsbG8hIEkgYW0gTm9uISBJdCdzIDI5LzExLzIwMjE=. You can decode it with Key.decode.

encode() → str#: Encode raw key with urlsafe_b64encode and add prefix.

hex() → str#: Returns key in hex representation

property key: bytes#: Returns key in raw

property key_type: int#: Returns current key type

property key_types: dict#: Returns all key types

class tgbox.keys.DirectoryKey(key: bytes)[source]#

Bases: Key

DirectoryKey is a Key that was added in the v1.3. In previous versions, FileKey was generated with the SHA256 over the MainKey and FileSalt. Now we will make it with the DirectoryKey. See Docs for more information.

Decodes Key by prefix and returns Key in one of Key classes.

B: BaseKey M: MainKey R: RequestKey S: ShareKey I: ImportKey F: FileKey E: EncryptedMainkey D: DirectoryKey H: HMACKey

Key example:: MSGVsbG8hIEkgYW0gTm9uISBJdCdzIDI5LzExLzIwMjE=. You can decode it with Key.decode.

encode() → str#: Encode raw key with urlsafe_b64encode and add prefix.

hex() → str#: Returns key in hex representation

property key: bytes#: Returns key in raw

property key_type: int#: Returns current key type

property key_types: dict#: Returns all key types

class tgbox.keys.EncryptedMainkey(key: bytes)[source]#

Bases: Key

This class represents encrypted mainkey. When you clone other’s RemoteBox we encrypt its MainKey with your BaseKey.

Decodes Key by prefix and returns Key in one of Key classes.

B: BaseKey M: MainKey R: RequestKey S: ShareKey I: ImportKey F: FileKey E: EncryptedMainkey D: DirectoryKey H: HMACKey

Key example:: MSGVsbG8hIEkgYW0gTm9uISBJdCdzIDI5LzExLzIwMjE=. You can decode it with Key.decode.

encode() → str#: Encode raw key with urlsafe_b64encode and add prefix.

hex() → str#: Returns key in hex representation

property key: bytes#: Returns key in raw

property key_type: int#: Returns current key type

property key_types: dict#: Returns all key types

class tgbox.keys.FileKey(key: bytes)[source]#

Bases: Key

FileKey is a key that used for encrypting file’s bytedata and its metadata on upload. The FileKey encrypts all of secret metadata values except the efile_path (encrypted file path), so user with which you share file from your Box will not know from which directory it was extracted.

Note

Usually you will not work with this class, API converts DirectoryKey to FileKey under the hood, but you can make it with tgbox.keys.make_filekey.

Decodes Key by prefix and returns Key in one of Key classes.

B: BaseKey M: MainKey R: RequestKey S: ShareKey I: ImportKey F: FileKey E: EncryptedMainkey D: DirectoryKey H: HMACKey

Key example:: MSGVsbG8hIEkgYW0gTm9uISBJdCdzIDI5LzExLzIwMjE=. You can decode it with Key.decode.

encode() → str#: Encode raw key with urlsafe_b64encode and add prefix.

hex() → str#: Returns key in hex representation

property key: bytes#: Returns key in raw

property key_type: int#: Returns current key type

property key_types: dict#: Returns all key types

class tgbox.keys.HMACKey(key: bytes)[source]#

Bases: Key

HMACKey is a Key that is used to make a HMAC of the bytestring. Typically, HMACKey is a result of a tgbox.keys.make_hmackey func.

Decodes Key by prefix and returns Key in one of Key classes.

B: BaseKey M: MainKey R: RequestKey S: ShareKey I: ImportKey F: FileKey E: EncryptedMainkey D: DirectoryKey H: HMACKey

Key example:: MSGVsbG8hIEkgYW0gTm9uISBJdCdzIDI5LzExLzIwMjE=. You can decode it with Key.decode.

encode() → str#: Encode raw key with urlsafe_b64encode and add prefix.

hex() → str#: Returns key in hex representation

property key: bytes#: Returns key in raw

property key_type: int#: Returns current key type

property key_types: dict#: Returns all key types

class tgbox.keys.ImportKey(key: bytes)[source]#

Bases: Key

The ImportKey is a key that Requester obtains after calling make_importkey function with the ShareKey. This is a decryption key for the requested object.

Run help(tgbox.keys.make_importkey) for information.

Decodes Key by prefix and returns Key in one of Key classes.

B: BaseKey M: MainKey R: RequestKey S: ShareKey I: ImportKey F: FileKey E: EncryptedMainkey D: DirectoryKey H: HMACKey

Key example:: MSGVsbG8hIEkgYW0gTm9uISBJdCdzIDI5LzExLzIwMjE=. You can decode it with Key.decode.

encode() → str#: Encode raw key with urlsafe_b64encode and add prefix.

hex() → str#: Returns key in hex representation

property key: bytes#: Returns key in raw

property key_type: int#: Returns current key type

property key_types: dict#: Returns all key types

class tgbox.keys.Key(key: bytes, key_type: int)[source]#

Bases: object

Metaclass that represents all keys.

Decodes Key by prefix and returns Key in one of Key classes.

B: BaseKey M: MainKey R: RequestKey S: ShareKey I: ImportKey F: FileKey E: EncryptedMainkey D: DirectoryKey H: HMACKey

Key example:: MSGVsbG8hIEkgYW0gTm9uISBJdCdzIDI5LzExLzIwMjE=. You can decode it with Key.decode.

encode() → str[source]#: Encode raw key with urlsafe_b64encode and add prefix.

hex() → str[source]#: Returns key in hex representation

property key: bytes#: Returns key in raw

property key_type: int#: Returns current key type

property key_types: dict#: Returns all key types

class tgbox.keys.MainKey(key: bytes)[source]#

Bases: Key

MainKey may be referred as “Box key”. This key encrypts all box data and used in FileKey creation. It’s one of your most important Key, as leakage of it will result in compromising all your encrypted files in RemoteBox & LocalBox.

When you clone other’s RemoteBox, Session data will be encrypted by BaseKey, not MainKey.

Usually you will see this Key as a result of keys.make_mainkey function.

Decodes Key by prefix and returns Key in one of Key classes.

B: BaseKey M: MainKey R: RequestKey S: ShareKey I: ImportKey F: FileKey E: EncryptedMainkey D: DirectoryKey H: HMACKey

Key example:: MSGVsbG8hIEkgYW0gTm9uISBJdCdzIDI5LzExLzIwMjE=. You can decode it with Key.decode.

encode() → str#: Encode raw key with urlsafe_b64encode and add prefix.

hex() → str#: Returns key in hex representation

property key: bytes#: Returns key in raw

property key_type: int#: Returns current key type

property key_types: dict#: Returns all key types

class tgbox.keys.Phrase(phrase: bytes | str)[source]#

Bases: object

This class represents passphrase

classmethod generate(words_count: int = 6) → Phrase[source]#

Generates passphrase

Arguments:

words_count (int, optional):: Words count in Phrase.

property phrase: bytes#: Returns current raw phrase

class tgbox.keys.RequestKey(key: bytes)[source]#

Bases: Key

The RequestKey is a key that Requester creates when [s]he wants to import Giver’s file, directory or even clone other’s RemoteBox and access all files.

With RequestKey Giver makes ShareKey. Run help(tgbox.keys.make_requestkey) for information.

Decodes Key by prefix and returns Key in one of Key classes.

B: BaseKey M: MainKey R: RequestKey S: ShareKey I: ImportKey F: FileKey E: EncryptedMainkey D: DirectoryKey H: HMACKey

Key example:: MSGVsbG8hIEkgYW0gTm9uISBJdCdzIDI5LzExLzIwMjE=. You can decode it with Key.decode.

encode() → str#: Encode raw key with urlsafe_b64encode and add prefix.

hex() → str#: Returns key in hex representation

property key: bytes#: Returns key in raw

property key_type: int#: Returns current key type

property key_types: dict#: Returns all key types

class tgbox.keys.ShareKey(key: bytes)[source]#

Bases: Key

The ShareKey is a key that Giver creates when [s]he wants to share file, directory or even the whole Box with the Requester.

With ShareKey Requester makes ImportKey. Run help(tgbox.keys.make_sharekey) for information.

Decodes Key by prefix and returns Key in one of Key classes.

B: BaseKey M: MainKey R: RequestKey S: ShareKey I: ImportKey F: FileKey E: EncryptedMainkey D: DirectoryKey H: HMACKey

Key example:: MSGVsbG8hIEkgYW0gTm9uISBJdCdzIDI5LzExLzIwMjE=. You can decode it with Key.decode.

encode() → str#: Encode raw key with urlsafe_b64encode and add prefix.

hex() → str#: Returns key in hex representation

property key: bytes#: Returns key in raw

property key_type: int#: Returns current key type

property key_types: dict#: Returns all key types

Function to create BaseKey. Uses the sha256(scrypt(...)).

Warning

RAM consumption is calculated by 128 * r * (n + p + 2).

Arguments:

phrase (bytes, Phrase):: Passphrase from which BaseKey will be created.
salt (bytes, int, optional):: Scrypt Salt.
n (int, optional):: Scrypt N.
r (int, optional):: Scrypt R.
p (int, optional):: Scrypt P.
dklen (int, optional):: Scrypt dklen.

tgbox.keys.make_dirkey(mainkey: MainKey, part_id: bytes) → DirectoryKey[source]#

Function to create DirectoryKey.

DirectoryKey is generated from the unique path PartID and the MainKey. We use the DirectoryKey to make a FileKey. See help(tgbox.keys.DirectoryKey) and docs for more information about this type of Key.

tgbox.keys.make_filekey(key: MainKey | DirectoryKey, file_salt: FileSalt) → FileKey[source]#

Function to create FileKey.

The FileKey is a Key that we use to encrypt the file and its metadata (except efile_path) on upload. Prior to the version 1.3 to make a FileKey we used MainKey and the FileSalt, which is randomly generated (on file preparation) 32 bytes. Started from now, instead of the MainKey we will use the DirectoryKey, but you can still generate old FileKey(s) with MainKey, it’s here only for backward compatibility and this is legacy.

MainKey or DirectoryKey can not be restored from the FileKey, so it’s safe-to-share.

The main benefit in using the DirectoryKey over MainKey is that in old versions you will need to share each of files from your Box separately, while now you can share the one DirectoryKey and Requester will be able to make all of the FileKeys to range of files in Dir by himself.

You still can share files separately, though.

See docs if you want to learn more about the Keys hierarchy structure & other things.

Arguments:

key (MainKey (legacy), DirectoryKey):: Key which will be used to make a FileKey.
file_salt (FileSalt):: FileSalt generated on file prepare.

tgbox.keys.make_hmackey(filekey: FileKey, file_salt: FileSalt) → HMACKey[source]#

Function to create HMACKey.

HMACKey is a Key that is used exclusively to derive a HMAC (by default SHA256) of a File or any target bytestring.

Arguments:

filekey (FileKey):: FileKey which will be used to make a HMACKey.
file_salt (FileSalt):: FileSalt that correspond to FileKey.

tgbox.keys.make_importkey(key: MainKey | BaseKey, sharekey: ShareKey, salt: FileSalt | BoxSalt | bytes | None = None) → ImportKey[source]#

Note

You may want to know what is RequestKey and ShareKey before using this. Use help() on another Key make functions.

ShareKey is a combination of encrypted by Alice (File/Main/Directory)Key and her pubkey. As Bob can create again RequestKey, which is PubKey of ECDH from sha256(key + salt) PrivKey, and already have PubKey of A, – B can create a shared secret, and decrypt A ShareKey to make an ImportKey.

Arguments:

key (MainKey, BaseKey):: Bob’s MainKey or BaseKey that was used on RequestKey creation.
sharekey (ShareKey):: Alice’s ShareKey.
salt (FileSalt, BoxSalt, bytes, optional):: Salt that was used on RequestKey creation.

tgbox.keys.make_mainkey(basekey: BaseKey, box_salt: BoxSalt) → MainKey[source]#

Function to create MainKey.

Arguments:

basekey (bytes):: Key which you received with scrypt function or any other key you want.
box_salt (BoxSalt):: BoxSalt generated on LocalBox creation.

tgbox.keys.make_requestkey(key: MainKey | BaseKey, salt: FileSalt | BoxSalt | bytes) → RequestKey[source]#

Function to create RequestKey.

All files in RemoteBox is encrypted with filekeys, so if you want to share (or import) file, then you need to get FileKey. For this purpose you can create RequestKey.

Alice has file in her Box which she wants to share with Bob. Then: A sends file to B. B forwards file to his Box, takes FileSalt from A File and MainKey of his Box and calls make_requestkey(key=mainkey, salt=file_salt).

RequestKey is a compressed pubkey of ECDH on SECP256K1 curve, B makes privkey with sha256(mainkey + salt) & exports pubkey to make a shared secret bytes (key, with which A will encrypt her filekey/mainkey. The encrypted (file/main)key is called ShareKey. Use help on make_sharekey.).

B sends received RequestKey to A. A makes ShareKey and sends it to B. B calls get_importkey and receives the ImportKey, which is, in fact, a FileKey.

No one except Alice and Bob will have FileKey. If Alice want to share entire Box (MainKey) with Bob, then Bob creates slightly different RequestKey with same function: make_requestkey(key=mainkey, salt=box_salt).

Please note that FileKey can only decrypt a some RemoteBox with which it is associated. However, if Alice will want to share the entire Directory of her Box files (i.e /home/alice/Pictures folder) then Bob can make a RequestKey to any file from this Directory, and Alice will make a ShareKey with a DirectoryKey instead of FileKey. See help on make_sharekey.

Note

Functions in this module is low-level, you can make RequestKey for a forwarded from A file by calling get_requestkey(...) method on EncryptedRemoteBoxFile | EncryptedRemoteBox.

Arguments:

key (MainKey, BaseKey):: Bob’s Key. If you want to import other’s file, then you need to specify here MainKey of your LocalBox, otherwise specify BaseKey (to clone RemoteBox)
salt (FileSalt, BoxSalt, bytes):: Most obvious salt is Alice’s BoxSalt or FileSalt, however, salt here is just some bytestring that will be hashed with the MainKey to make the output ECDH keys unique, so you can specify here any bytes value if you understand consequences (you will need to reuse it on make_importkey).

Function to create ShareKey.

Note

You may want to know what is RequestKey before reading this. Please, run help on make_requestkey to get info.

Alice received RequestKey from Bob. But what she should do next? As reqkey is just EC-pubkey, she wants to make a shared secret key. A makes her own privkey, with sha256(mainkey + sha256(salt + requestkey)) & initializes ECDH with B pubkey and her privkey. After this, A makes a hashed with SHA256 shared secret, which will be used as 32-byte length AES-CBC key & encrypts her File|Main|Directory key. IV here is first 16 bytes of the sha256(requestkey). After, she prepends her pubkey to the resulted encrypted File|Main|Directory key and sends it to Bob.

With A pubkey, B can easily get the same shared secret and decrypt ShareKey to make the ImportKey.

The things will be much less complicated if Alice don’t mind to share her File, Dir or Box with ALL peoples. In this case we don’t even need to make a ShareKey, ImportKey will be returned from the raw target Key.

Arguments:

key (MainKey, FileKey, DirectoryKey):

o If key is instance of MainKey: Box key. Specify only this kwarg and ignore requestkey if you want to share your Box with ALL peoples. Your Box key – MainKey will be NOT encrypted.

o If key is instance of FileKey: File key. Specify only this kwarg if you want to share your File with ALL peoples. No encryption if RequestKey (as requestkey) is not specified.

o If key is instance of DirectoryKey: Dir key. Specify only this kwarg if you want to share your File with ALL peoples. No encryption if RequestKey (as requestkey) is not specified.

salt (FileSalt, BoxSalt, bytes, optional):

Most obvious salt is Alice’s BoxSalt or FileSalt, however, salt here is just some bytestring that will be hashed with the MainKey to make the output ECDH keys unique, so you can specify here any bytes value if you understand consequences. For example, we will use PartID (bytes) as salt on DirectoryKey sharing.

requestkey (RequestKey, optional):

RequestKey of Bob. With this must be specified salt.

tgbox.tools module#

This module stores utils required by API.

class tgbox.tools.OpenPretender(flo: BinaryIO, aes_state: tgbox.crypto.AESwState, hmac_state: hashlib.HMAC, file_size: int)[source]#

Bases: object

Class to wrap Tgbox AES Generators and make it look like opened to “rb”-read file.

close() → None[source]#

concat_metadata(metadata: bytes) → None[source]#: Concates metadata to the file as (metadata + file).

get_expected_size()[source]#: Returns expected actual Telegram document size after upload. We use it in push_file()

async read(size: int = -1) → bytes[source]#

Returns size bytes from async Generator.

This method is async only because we use File uploading from the async library. You can use tgbox.sync in your sync code for reading.

Arguments:

size (int):: Amount of bytes to return. By default is negative (return all). Must be divisible by 16, and >= 64.

seekable(*args, **kwargs) → bool[source]#

tell() → int[source]#

class tgbox.tools.PackedAttributes[source]#

Bases: object

This class is used to pack items to bytestring. We use it to pack file metadata as well as pack a user’s custom attributes (cattrs), then encrypt it and attach to RBFile.

We store key/value length in 3 bytes, so the max key/value length is 256^3-1.

<key-length>key<value-length>value<…>

Every key/value will be randomly shuffled on the packing process.

static pack(*, random_seed: bytes | None = None, protected_keys: tuple | None = None, **kwargs) → bytes[source]#

Will make bytestring from your kwargs. Any kwarg always must be bytes.

pack(x=5) will not work; pack(x=b'') is correct.

We shuffle all key/value before packing, so you can specify random_seed. Otherwise, urandom(32) will be used instead.

Keys specified in the protect_key tuple will never be in the start or or in the end of the packed bytestring.

protect_key is for internal usage.

static unpack(pattr: bytes) → Dict[str, bytes][source]#

Will parse PackedAttributes.pack bytestring and convert it to the python dictionary.

Every PackedAttributes bytestring must contain 0xFF as first byte. If not, or if error, will return {}.

class tgbox.tools.SearchFilter(**kwargs)[source]#

Bases: object

Container that filters search in DecryptedRemoteBox or DecryptedLocalBox.

The SearchFilter has two filters: the Include and Exclude. On search, all matching to include files will be yielded, while all matching to exclude will be not yielded (ignored).

The tgbox.tools.search_generator will firstly check for include filters, so its priority higher.
The tgbox.tools.search_generator will yield files that match all of filters, not one of it.
The SearchFilter accepts list as kwargs value. You can SearchFilter(id=[3,5,10]).
The SearchFilter(**kwargs) will add all filters to the include. Also use SearchFilter.include(...) & SearchFilter.exclude(...) methods after initialization.

All filters:

scope str: Define a path as search scope:
The scope is an absolute directory in which we will search your file by other filters. By default, the tgbox.api.utils.search_generator will search over the entire LocalBox. This can be slow if you’re have too many files.

Example: let’s imagine that You’re a Linux user which share it’s Box with the Windows user. In this case, Your LocalBox will contain a path parts on the '/' (Linux) and 'C:\' (Windows) roots. If You know that some file was uploaded by Your friend, then You can specify a scope='C:\' to ignore all files uploaded from the Linux machine. This will significantly fasten the search process, because almost all filters require to select row from the LocalBox DB, decrypt Metadata and compare its values with ones from SearchFilter.

!: The scope will be ignored on RemoteBox search.

!: The min_id & max_id will be ignored if scope used.
id integer: File ID
cattrs dict: File CustomAttributes:
To search for CATTRS you need to specify a dict.

E.g: If file cattrs={b'comment': b'hi!'}, then filter cattrs={b'comment': b'h'} will match.

By default, tgbox.api.utils.search_generator will use an in, like b'h' in b'hi!', but you can set a re flag to use regular expressions, so filter cattrs={b'comment': b'hi(.)'} will match.
file_path pathlib.Path, str
file_name str: File name
file_salt bytes/str: File salt
verbyte bytes: File version byte
mime str: File mime type
sender integer/str: File sender name or ID:
Only works on RemoteBox files and only on RemoteBox that enabled Sign Messages (& Show Author Profiles, optionally for IDs), otherwise will never return files.

This filter can be Sender name or Sender ID, e.g sender=’black sabbath’ or sender=36265675; if ‘sender’ is string but .isnumeric(), Generator will also convert it to int and check against ID (if available), essentially, sender=36265675 and sender=’36265675’ both valid, but latter will also check ‘36265675’ in name.
minor_version integer: File minor version
min_id integer: File ID should be > min_id
max_id integer: File ID should be < max_id
min_size integer: File Size should be > min_size
max_size integer: File Size should be < max_size
min_time integer/float: Upload Time should be > min_time
max_time integer/float: Upload Time should be < max_time
re bool: re_search for every bytes
imported bool: Yield only imported files
non_recursive_scope bool: Disable recursive scope search

exclude(**kwargs) → SearchFilter[source]#: Will extend excluded filters

include(**kwargs) → SearchFilter[source]#: Will extend included filters

async tgbox.tools.anext(aiterator, default=Ellipsis)[source]#: Analogue to Python 3.10+ anext()

tgbox.tools.bytes_to_int(bytes_: bytes, signed: bool | None = False) → int[source]#: Converts bytes to int with Big byteorder.

async tgbox.tools.get_media_duration(file_path: str) → int[source]#: Returns video/audio duration with ffmpeg in seconds.

tgbox.tools.guess_path_type(path: str | Path) → str[source]#

This function will try to guess file path type. It can be Windows-like or Unix-like

Returns ‘windows’ or ‘unix’

tgbox.tools.install_ffmpeg(ffmpeg_executable: str | Path) → Path[source]#

Use this function to “install” your custom FFmpeg executable within the tgbox Protocol. We will copy your ffmpeg_executable to the tgbox/other directory. Protocol will always use it, even if ffmpeg is installed on your machine.

Arguments:

ffmpeg_executable (Union[str, Path]):

Path to ffmpeg executable. Please note that its name should start with “ffmpeg”. If you’re on Windows, “.exe” suffix must be present.

We will always lowercase name of your executable when copying to tgbox/other.

tgbox.tools.int_to_bytes(int_: int, length: int | None = None, signed: bool | None = False) → bytes[source]#: Converts int to bytes with Big byteorder.

tgbox.tools.make_file_fingerprint(mainkey: MainKey, file_path: str | Path) → bytes[source]#

Function to make a file Fingerprint.

Fingerprint is a SHA256 over mainkey and file_path parts, not a hash of a file. We use it to check if prepared file is unique or not (and raise error if Box already have some file with the same name and path).

Arguments:

mainkey (MainKey):

The MainKey of your Box

file_path (Union[str, Path]):

A file path from which we will make a Fingerprint. It should include a file name!

/home/user/ (directory) is NOT OK! /home/user/file.txt (file) is OK!

tgbox.tools.make_general_path(path: str | Path) → Path[source]#: This function will make a valid UNIX-like Path from the Windows-like on the UNIX-like systems.

async tgbox.tools.make_media_preview(file_path: PathLike, x: int = 128, y: int = -1) → BinaryIO[source]#: Makes x:y sized thumbnail of the video/audio with ffmpeg. “-1” preserves one of side size.

tgbox.tools.make_safe_file_path(path: str | Path) → Path[source]#

This function will make a safe file path (a file path that can be easily inserted into another path). This is mostly for internal purposes, i.e DecryptedRemoteBox.download()

This function will make a: @/home/non/test from /home/non/test CUsersnontest from C:Usersnontest

…so this path can be easily inserted into another, i.e DownloadsTGBOX/@/home/non/test

path must be absolute.

tgbox.tools.ppart_id_generator(path: Path, mainkey: MainKey) → Generator[tuple, None, None][source]#

This generator will iterate over path parts and yield their unique IDs. We will use this to better navigate over abstract Folders in the LocalBox.

The path shouldn’t contain a file name, otherwise directory will contain it as folder.

/home/user/ is OK /home/user/file.txt is NOT

Will yield a tuple (PART, PARENT_PART_ID, PART_ID)

tgbox.tools.prbg(size: int) → bytes[source]#: Will generate size pseudo-random bytes.

tgbox.defaults module#

This module stores API defaults.

class tgbox.defaults.Limits(*values)[source]#

Bases: IntEnum

Default TGBOX API limits

FILE_PATH_MAX: int = 4096#

METADATA_MAX: int = 1000000#

as_integer_ratio()#

Return a pair of integers, whose ratio is equal to the original int.

The ratio is in lowest terms and has a positive denominator.

>>> (10).as_integer_ratio()
(10, 1)
>>> (-10).as_integer_ratio()
(-10, 1)
>>> (0).as_integer_ratio()
(0, 1)

bit_count()#

Number of ones in the binary representation of the absolute value of self.

Also known as the population count.

>>> bin(13)
'0b1101'
>>> (13).bit_count()
3

bit_length()#

Number of bits necessary to represent self in binary.

>>> bin(37)
'0b100101'
>>> (37).bit_length()
6

conjugate()#: Returns self, the complex conjugate of any int.

denominator#: the denominator of a rational number in lowest terms

classmethod from_bytes(bytes, byteorder='big', *, signed=False)#

Return the integer represented by the given array of bytes.

bytes: Holds the array of bytes to convert. The argument must either support the buffer protocol or be an iterable object producing bytes. Bytes and bytearray are examples of built-in objects that support the buffer protocol.
byteorder: The byte order used to represent the integer. If byteorder is ‘big’, the most significant byte is at the beginning of the byte array. If byteorder is ‘little’, the most significant byte is at the end of the byte array. To request the native byte order of the host system, use `sys.byteorder’ as the byte order value. Default is to use ‘big’.
signed: Indicates whether two’s complement is used to represent the integer.

imag#: the imaginary part of a complex number

is_integer()#: Returns True. Exists for duck type compatibility with float.is_integer.

numerator#: the numerator of a rational number in lowest terms

real#: the real part of a complex number

to_bytes(length=1, byteorder='big', *, signed=False)#

Return an array of bytes representing an integer.

length: Length of bytes object to use. An OverflowError is raised if the integer is not representable with the given number of bytes. Default is length 1.
byteorder: The byte order used to represent the integer. If byteorder is ‘big’, the most significant byte is at the beginning of the byte array. If byteorder is ‘little’, the most significant byte is at the end of the byte array. To request the native byte order of the host system, use `sys.byteorder’ as the byte order value. Default is to use ‘big’.
signed: Determines whether two’s complement is used to represent the integer. If signed is False and a negative integer is given, an OverflowError is raised.

class tgbox.defaults.Scrypt(*values)[source]#

Bases: IntEnum

Default Scrypt KDF configuration

DKLEN: int = 32#

N: int = 1048576#

P: int = 1#

R: int = 8#

SALT: int = 25241879641352434918417594454429676564939529791904169407552470269367179288935#

as_integer_ratio()#

Return a pair of integers, whose ratio is equal to the original int.

The ratio is in lowest terms and has a positive denominator.

>>> (10).as_integer_ratio()
(10, 1)
>>> (-10).as_integer_ratio()
(-10, 1)
>>> (0).as_integer_ratio()
(0, 1)

bit_count()#

Number of ones in the binary representation of the absolute value of self.

Also known as the population count.

>>> bin(13)
'0b1101'
>>> (13).bit_count()
3

bit_length()#

Number of bits necessary to represent self in binary.

>>> bin(37)
'0b100101'
>>> (37).bit_length()
6

conjugate()#: Returns self, the complex conjugate of any int.

denominator#: the denominator of a rational number in lowest terms

classmethod from_bytes(bytes, byteorder='big', *, signed=False)#

Return the integer represented by the given array of bytes.

bytes: Holds the array of bytes to convert. The argument must either support the buffer protocol or be an iterable object producing bytes. Bytes and bytearray are examples of built-in objects that support the buffer protocol.
byteorder: The byte order used to represent the integer. If byteorder is ‘big’, the most significant byte is at the beginning of the byte array. If byteorder is ‘little’, the most significant byte is at the end of the byte array. To request the native byte order of the host system, use `sys.byteorder’ as the byte order value. Default is to use ‘big’.
signed: Indicates whether two’s complement is used to represent the integer.

imag#: the imaginary part of a complex number

is_integer()#: Returns True. Exists for duck type compatibility with float.is_integer.

numerator#: the numerator of a rational number in lowest terms

real#: the real part of a complex number

to_bytes(length=1, byteorder='big', *, signed=False)#

Return an array of bytes representing an integer.

length: Length of bytes object to use. An OverflowError is raised if the integer is not representable with the given number of bytes. Default is length 1.
byteorder: The byte order used to represent the integer. If byteorder is ‘big’, the most significant byte is at the beginning of the byte array. If byteorder is ‘little’, the most significant byte is at the end of the byte array. To request the native byte order of the host system, use `sys.byteorder’ as the byte order value. Default is to use ‘big’.
signed: Determines whether two’s complement is used to represent the integer. If signed is False and a negative integer is given, an OverflowError is raised.

class tgbox.defaults.UploadLimits(*values)[source]#

Bases: IntEnum

Telegram filesize limits

DEFAULT = 2000000000#

PREMIUM = 4000000000#

as_integer_ratio()#

Return a pair of integers, whose ratio is equal to the original int.

The ratio is in lowest terms and has a positive denominator.

>>> (10).as_integer_ratio()
(10, 1)
>>> (-10).as_integer_ratio()
(-10, 1)
>>> (0).as_integer_ratio()
(0, 1)

bit_count()#

Number of ones in the binary representation of the absolute value of self.

Also known as the population count.

>>> bin(13)
'0b1101'
>>> (13).bit_count()
3

bit_length()#

Number of bits necessary to represent self in binary.

>>> bin(37)
'0b100101'
>>> (37).bit_length()
6

conjugate()#: Returns self, the complex conjugate of any int.

denominator#: the denominator of a rational number in lowest terms

classmethod from_bytes(bytes, byteorder='big', *, signed=False)#

Return the integer represented by the given array of bytes.

bytes: Holds the array of bytes to convert. The argument must either support the buffer protocol or be an iterable object producing bytes. Bytes and bytearray are examples of built-in objects that support the buffer protocol.
byteorder: The byte order used to represent the integer. If byteorder is ‘big’, the most significant byte is at the beginning of the byte array. If byteorder is ‘little’, the most significant byte is at the end of the byte array. To request the native byte order of the host system, use `sys.byteorder’ as the byte order value. Default is to use ‘big’.
signed: Indicates whether two’s complement is used to represent the integer.

imag#: the imaginary part of a complex number

is_integer()#: Returns True. Exists for duck type compatibility with float.is_integer.

numerator#: the numerator of a rational number in lowest terms

real#: the real part of a complex number

to_bytes(length=1, byteorder='big', *, signed=False)#

Return an array of bytes representing an integer.

length: Length of bytes object to use. An OverflowError is raised if the integer is not representable with the given number of bytes. Default is length 1.
byteorder: The byte order used to represent the integer. If byteorder is ‘big’, the most significant byte is at the beginning of the byte array. If byteorder is ‘little’, the most significant byte is at the end of the byte array. To request the native byte order of the host system, use `sys.byteorder’ as the byte order value. Default is to use ‘big’.
signed: Determines whether two’s complement is used to represent the integer. If signed is False and a negative integer is given, an OverflowError is raised.

tgbox package

Contents

tgbox package#

tgbox.api.local module#

tgbox.api.remote module#

tgbox.api.db module#

tgbox.api.utils module#

tgbox.api.abstract module#

tgbox.api.sync module#

tgbox.crypto module#

tgbox.errors module#

tgbox.fastelethon module#

tgbox.keys module#

tgbox.tools module#

tgbox.defaults module#