iframe-proxy | Sunbelt Computer Software

History

226 lines (179 loc) · 6.54 KB

Raw

100

101

102

103

104

105

106

107

108

109

110

111

112

113

114

115

116

117

118

119

120

121

122

123

124

125

126

127

128

129

130

131

132

133

134

135

136

137

138

139

140

141

142

143

144

145

146

147

148

149

150

151

152

153

154

155

156

157

158

159

160

161

162

163

164

165

166

167

168

169

170

171

172

173

174

175

176

177

178

179

180

181

182

183

184

185

186

187

188

189

190

191

192

193

194

195

196

197

198

199

200

201

202

203

204

205

206

207

208

209

210

211

212

213

214

215

216

217

218

219

220

221

222

223

224

225

226

# Hydrogram - Telegram MTProto API Client Library for Python

# This file is part of Hydrogram.

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

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

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

# (at your option) any later version.

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

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

# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the

# GNU Lesser General Public License for more details.

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

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

from __future__ import annotations

import base64

import struct

from abc import ABC, abstractmethod

from typing import Union

from hydrogram import raw

InputPeer = Union[raw.types.InputPeerUser, raw.types.InputPeerChat, raw.types.InputPeerChannel]

class BaseStorage(ABC):

"""The BaseStorage class is an abstract base class defining the interface

for different storage engines used by Hyrogram.

Parameters:

name (``str``):

The name of the session.

"""

SESSION_STRING_FORMAT: str = ">BI?256sQ?"

def __init__(self, name: str) -> None:

self.name = name

@abstractmethod

async def open(self) -> None:

"""Opens the storage engine."""

...

@abstractmethod

async def save(self) -> None:

"""Saves the current state of the storage engine."""

...

@abstractmethod

async def close(self) -> None:

"""Closes the storage engine."""

...

@abstractmethod

async def delete(self) -> None:

"""Deletes the storage."""

...

@abstractmethod

async def update_peers(self, peers: list[tuple[int, int, str, str, str]]) -> None:

"""Update the peers table with the provided information.

Parameters:

peers (``List[Tuple[int, int, str, str, str]]``): A list of tuples containing the

information of the peers to be updated. Each tuple must contain:

- ``int``: The peer id.

- ``int``: The peer access hash.

- ``str``: The peer type (user, chat, or channel).

- ``str``: The peer username (if any).

- ``str``: The peer phone number (if any).

"""

...

@abstractmethod

async def get_peer_by_id(self, peer_id: int) -> InputPeer:

"""Retrieve a peer by its ID.

Parameters:

peer_id (``int``):

The ID of the peer to retrieve.

Returns:

:obj:`~hydrogram.storage.base.InputPeer`: The retrieved peer.

"""

...

@abstractmethod

async def get_peer_by_username(self, username: str) -> InputPeer:

"""Retrieve a peer by its username.

Parameters:

username (``str``):

The username of the peer to retrieve.

Returns:

:obj:`~hydrogram.storage.base.InputPeer`: The retrieved peer.

"""

...

@abstractmethod

async def get_peer_by_phone_number(self, phone_number: str) -> InputPeer:

"""Retrieve a peer by its phone number.

Parameters:

phone_number (``str``):

The phone number of the peer to retrieve.

Returns:

:obj:`~hydrogram.storage.base.InputPeer`: The retrieved peer.

"""

...

@abstractmethod

async def dc_id(self, value: int | None = None) -> int:

"""Get or set the DC ID of the current session.

Parameters:

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

The DC ID to set.

Returns:

``int``: The current DC ID if no value is provided.

"""

...

@abstractmethod

async def api_id(self, value: int | None = None) -> int:

"""Get or set the API ID of the current session.

Parameters:

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

The API ID to set.

Returns:

``int``: The current API ID if no value is provided.

"""

...

@abstractmethod

async def test_mode(self, value: bool | None = None) -> bool:

"""Get or set the test mode of the current session.

Parameters:

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

The test mode to set.

Returns:

``bool``: The current test mode if no value is provided.

"""

...

@abstractmethod

async def auth_key(self, value: bytes | None = None) -> bytes:

"""Get or set the authorization key of the current session.

Parameters:

value (``bytes``, *optional*):

The authorization key to set.

Returns:

``bytes``: The current authorization key if no value is provided.

"""

...

@abstractmethod

async def date(self, value: int | None = None) -> int:

"""Get or set the date of the current session.

Parameters:

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

The date to set.

Returns:

``int``: The current date if no value is provided.

"""

...

@abstractmethod

async def user_id(self, value: int | None = None) -> int:

"""Get or set the user ID of the current session.

Parameters:

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

The user ID to set.

Returns:

``int``: The current user ID if no value is provided.

"""

...

@abstractmethod

async def is_bot(self, value: bool | None = None) -> bool:

"""Get or set the bot flag of the current session.

Parameters:

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

The bot flag to set.

Returns:

``bool``: The current bot flag if no value is provided.

"""

...

async def export_session_string(self) -> str:

"""Exports the session string for the current session.

Returns:

``str``: The session string for the current session.

"""

packed = struct.pack(

self.SESSION_STRING_FORMAT,

await self.dc_id(),

await self.api_id(),

await self.test_mode(),

await self.auth_key(),

await self.user_id(),

await self.is_bot(),

)

return base64.urlsafe_b64encode(packed).decode().rstrip("=")

Sunbelt Computer Software

PL/B Language Development and Support

Navigation Menu

Uh oh!

Uh oh!

Sunbelt Computer Software

PL/B Language Development and Support

Uh oh!

Uh oh!

FilesExpand file tree

base.py

Latest commit

History

base.py

File metadata and controls