mirror of
https://github.com/LonamiWebs/Telethon.git
synced 2025-11-15 21:30:45 +00:00
Make lint happier
This commit is contained in:
@@ -26,7 +26,8 @@ class BinaryReader:
|
||||
|
||||
# region Reading
|
||||
|
||||
# "All numbers are written as little endian." |> Source: https://core.telegram.org/mtproto
|
||||
# "All numbers are written as little endian."
|
||||
# https://core.telegram.org/mtproto
|
||||
def read_byte(self):
|
||||
"""Reads a single byte value"""
|
||||
return self.read(1)[0]
|
||||
@@ -56,8 +57,7 @@ class BinaryReader:
|
||||
"""Read the given amount of bytes"""
|
||||
result = self.reader.read(length)
|
||||
if len(result) != length:
|
||||
raise BufferError(
|
||||
'Trying to read outside the data bounds (no more data left to read)')
|
||||
raise BufferError('No more data left to read')
|
||||
|
||||
return result
|
||||
|
||||
@@ -70,7 +70,9 @@ class BinaryReader:
|
||||
# region Telegram custom reading
|
||||
|
||||
def tgread_bytes(self):
|
||||
"""Reads a Telegram-encoded byte array, without the need of specifying its length"""
|
||||
"""Reads a Telegram-encoded byte array,
|
||||
without the need of specifying its length
|
||||
"""
|
||||
first_byte = self.read_byte()
|
||||
if first_byte == 254:
|
||||
length = self.read_byte() | (self.read_byte() << 8) | (
|
||||
@@ -102,7 +104,9 @@ class BinaryReader:
|
||||
raise ValueError('Invalid boolean code {}'.format(hex(value)))
|
||||
|
||||
def tgread_date(self):
|
||||
"""Reads and converts Unix time (used by Telegram) into a Python datetime object"""
|
||||
"""Reads and converts Unix time (used by Telegram)
|
||||
into a Python datetime object
|
||||
"""
|
||||
value = self.read_int()
|
||||
return None if value == 0 else datetime.fromtimestamp(value)
|
||||
|
||||
@@ -152,7 +156,9 @@ class BinaryReader:
|
||||
self.reader.seek(position)
|
||||
|
||||
def seek(self, offset):
|
||||
"""Seeks the stream position given an offset from the current position. May be negative"""
|
||||
"""Seeks the stream position given an offset from the
|
||||
current position. The offset may be negative
|
||||
"""
|
||||
self.reader.seek(offset, os.SEEK_CUR)
|
||||
|
||||
# endregion
|
||||
|
||||
@@ -22,21 +22,22 @@ class BinaryWriter:
|
||||
|
||||
# region Writing
|
||||
|
||||
# "All numbers are written as little endian." |> Source: https://core.telegram.org/mtproto
|
||||
# "All numbers are written as little endian."
|
||||
# https://core.telegram.org/mtproto
|
||||
def write_byte(self, value):
|
||||
"""Writes a single byte value"""
|
||||
self.writer.write(pack('B', value))
|
||||
self.written_count += 1
|
||||
|
||||
def write_int(self, value, signed=True):
|
||||
"""Writes an integer value (4 bytes), which can or cannot be signed"""
|
||||
"""Writes an integer value (4 bytes), optionally signed"""
|
||||
self.writer.write(
|
||||
int.to_bytes(
|
||||
value, length=4, byteorder='little', signed=signed))
|
||||
self.written_count += 4
|
||||
|
||||
def write_long(self, value, signed=True):
|
||||
"""Writes a long integer value (8 bytes), which can or cannot be signed"""
|
||||
"""Writes a long integer value (8 bytes), optionally signed"""
|
||||
self.writer.write(
|
||||
int.to_bytes(
|
||||
value, length=8, byteorder='little', signed=signed))
|
||||
@@ -101,7 +102,9 @@ class BinaryWriter:
|
||||
self.write_int(0x997275b5 if boolean else 0xbc799737, signed=False)
|
||||
|
||||
def tgwrite_date(self, datetime):
|
||||
"""Converts a Python datetime object into Unix time (used by Telegram) and writes it"""
|
||||
"""Converts a Python datetime object into Unix time
|
||||
(used by Telegram) and writes it
|
||||
"""
|
||||
value = 0 if datetime is None else int(datetime.timestamp())
|
||||
self.write_int(value)
|
||||
|
||||
@@ -127,14 +130,18 @@ class BinaryWriter:
|
||||
self.writer.close()
|
||||
|
||||
def get_bytes(self, flush=True):
|
||||
"""Get the current bytes array content from the buffer, optionally flushing first"""
|
||||
"""Get the current bytes array content from the buffer,
|
||||
optionally flushing first
|
||||
"""
|
||||
if flush:
|
||||
self.writer.flush()
|
||||
return self.writer.raw.getvalue()
|
||||
|
||||
def get_written_bytes_count(self):
|
||||
"""Gets the count of bytes written in the buffer.
|
||||
This may NOT be equal to the stream length if one was provided when initializing the writer"""
|
||||
This may NOT be equal to the stream length if one
|
||||
was provided when initializing the writer
|
||||
"""
|
||||
return self.written_count
|
||||
|
||||
# with block
|
||||
|
||||
Reference in New Issue
Block a user