python/aiogram
1
0
Fork 0
mirror of https://github.com/aiogram/aiogram.git synced 2026-04-08 16:37:47 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

updated typing notations

Browse source
This commit is contained in:
Mykola Soloduha 2020-07-30 14:06:51 +03:00
parent d1452b1620
commit 6161fd0a69
5 changed files with 44 additions and 44 deletions
Download patch file Download diff file Expand all files Collapse all files

18
aiogram/contrib/fsm_storage/memory.py
View file

@ -35,19 +35,19 @@ class MemoryStorage(BaseStorage):
user: typing.Union[str, int, None] = None,
default: typing.Optional[str] = None) -> typing.Optional[str]:
chat, user = self.resolve_address(chat=chat, user=user)
return self.data[chat][user]['state']
return self.data[chat][user]['state'] or default
async def get_data(self, *,
chat: typing.Union[str, int, None] = None,
user: typing.Union[str, int, None] = None,
default: typing.Optional[str] = None) -> typing.Dict:
default: typing.Optional[dict] = None) -> dict:
chat, user = self.resolve_address(chat=chat, user=user)
return copy.deepcopy(self.data[chat][user]['data'])
return copy.deepcopy(self.data[chat][user]['data']) or default or {}
async def update_data(self, *,
chat: typing.Union[str, int, None] = None,
user: typing.Union[str, int, None] = None,
data: typing.Dict = None, **kwargs):
data: typing.Optional[dict] = None, **kwargs):
if data is None:
data = {}
chat, user = self.resolve_address(chat=chat, user=user)
@ -63,7 +63,7 @@ class MemoryStorage(BaseStorage):
async def set_data(self, *,
chat: typing.Union[str, int, None] = None,
user: typing.Union[str, int, None] = None,
data: typing.Dict = None):
data: typing.Optional[dict] = None):
chat, user = self.resolve_address(chat=chat, user=user)
self.data[chat][user]['data'] = copy.deepcopy(data)
@ -81,21 +81,21 @@ class MemoryStorage(BaseStorage):
async def get_bucket(self, *,
chat: typing.Union[str, int, None] = None,
user: typing.Union[str, int, None] = None,
default: typing.Optional[dict] = None) -> typing.Dict:
default: typing.Optional[dict] = None) -> dict:
chat, user = self.resolve_address(chat=chat, user=user)
return copy.deepcopy(self.data[chat][user]['bucket'])
return copy.deepcopy(self.data[chat][user]['bucket']) or default or {}
async def set_bucket(self, *,
chat: typing.Union[str, int, None] = None,
user: typing.Union[str, int, None] = None,
bucket: typing.Dict = None):
bucket: typing.Optional[dict] = None):
chat, user = self.resolve_address(chat=chat, user=user)
self.data[chat][user]['bucket'] = copy.deepcopy(bucket)
async def update_bucket(self, *,
chat: typing.Union[str, int, None] = None,
user: typing.Union[str, int, None] = None,
bucket: typing.Dict = None, **kwargs):
bucket: typing.Optional[dict] = None, **kwargs):
if bucket is None:
bucket = {}
chat, user = self.resolve_address(chat=chat, user=user)

30
aiogram/contrib/fsm_storage/redis.py
View file

@ -73,7 +73,7 @@ class RedisStorage(BaseStorage):
async def get_record(self, *,
chat: typing.Union[str, int, None] = None,
user: typing.Union[str, int, None] = None) -> typing.Dict:
user: typing.Union[str, int, None] = None) -> dict:
"""
Get record from storage
@ -121,9 +121,9 @@ class RedisStorage(BaseStorage):
return record['state']
async def get_data(self, *, chat: typing.Union[str, int, None] = None, user: typing.Union[str, int, None] = None,
default: typing.Optional[str] = None) -> typing.Dict:
default: typing.Optional[dict] = None) -> dict:
record = await self.get_record(chat=chat, user=user)
return record['data']
return record['data'] or default or {}
async def set_state(self, *, chat: typing.Union[str, int, None] = None, user: typing.Union[str, int, None] = None,
state: typing.Optional[typing.AnyStr] = None):
@ -131,12 +131,12 @@ class RedisStorage(BaseStorage):
await self.set_record(chat=chat, user=user, state=state, data=record['data'])
async def set_data(self, *, chat: typing.Union[str, int, None] = None, user: typing.Union[str, int, None] = None,
data: typing.Dict = None):
data: typing.Optional[dict] = None):
record = await self.get_record(chat=chat, user=user)
await self.set_record(chat=chat, user=user, state=record['state'], data=data)
async def update_data(self, *, chat: typing.Union[str, int, None] = None, user: typing.Union[str, int, None] = None,
data: typing.Dict = None, **kwargs):
data: typing.Optional[dict] = None, **kwargs):
if data is None:
data = {}
record = await self.get_record(chat=chat, user=user)
@ -179,18 +179,18 @@ class RedisStorage(BaseStorage):
return True
async def get_bucket(self, *, chat: typing.Union[str, int, None] = None, user: typing.Union[str, int, None] = None,
default: typing.Optional[str] = None) -> typing.Dict:
default: typing.Optional[dict] = None) -> dict:
record = await self.get_record(chat=chat, user=user)
return record.get('bucket', {})
return record.get('bucket') or default or {}
async def set_bucket(self, *, chat: typing.Union[str, int, None] = None, user: typing.Union[str, int, None] = None,
bucket: typing.Dict = None):
bucket: typing.Optional[dict] = None):
record = await self.get_record(chat=chat, user=user)
await self.set_record(chat=chat, user=user, state=record['state'], data=record['data'], bucket=bucket)
async def update_bucket(self, *, chat: typing.Union[str, int, None] = None,
user: typing.Union[str, int, None] = None,
bucket: typing.Dict = None, **kwargs):
bucket: typing.Optional[dict] = None, **kwargs):
record = await self.get_record(chat=chat, user=user)
record_bucket = record.get('bucket', {})
if bucket is None:
@ -277,7 +277,7 @@ class RedisStorage2(BaseStorage):
return await redis.get(key, encoding='utf8') or None
async def get_data(self, *, chat: typing.Union[str, int, None] = None, user: typing.Union[str, int, None] = None,
default: typing.Optional[dict] = None) -> typing.Dict:
default: typing.Optional[dict] = None) -> dict:
chat, user = self.check_address(chat=chat, user=user)
key = self.generate_key(chat, user, STATE_DATA_KEY)
redis = await self.redis()
@ -297,14 +297,14 @@ class RedisStorage2(BaseStorage):
await redis.set(key, state, expire=self._state_ttl)
async def set_data(self, *, chat: typing.Union[str, int, None] = None, user: typing.Union[str, int, None] = None,
data: typing.Dict = None):
data: typing.Optional[dict] = None):
chat, user = self.check_address(chat=chat, user=user)
key = self.generate_key(chat, user, STATE_DATA_KEY)
redis = await self.redis()
await redis.set(key, json.dumps(data), expire=self._data_ttl)
async def update_data(self, *, chat: typing.Union[str, int, None] = None, user: typing.Union[str, int, None] = None,
data: typing.Dict = None, **kwargs):
data: typing.Optional[dict] = None, **kwargs):
if data is None:
data = {}
temp_data = await self.get_data(chat=chat, user=user, default={})
@ -315,7 +315,7 @@ class RedisStorage2(BaseStorage):
return True
async def get_bucket(self, *, chat: typing.Union[str, int, None] = None, user: typing.Union[str, int, None] = None,
default: typing.Optional[dict] = None) -> typing.Dict:
default: typing.Optional[dict] = None) -> dict:
chat, user = self.check_address(chat=chat, user=user)
key = self.generate_key(chat, user, STATE_BUCKET_KEY)
redis = await self.redis()
@ -325,7 +325,7 @@ class RedisStorage2(BaseStorage):
return default or {}
async def set_bucket(self, *, chat: typing.Union[str, int, None] = None, user: typing.Union[str, int, None] = None,
bucket: typing.Dict = None):
bucket: typing.Optional[dict] = None):
chat, user = self.check_address(chat=chat, user=user)
key = self.generate_key(chat, user, STATE_BUCKET_KEY)
redis = await self.redis()
@ -333,7 +333,7 @@ class RedisStorage2(BaseStorage):
async def update_bucket(self, *, chat: typing.Union[str, int, None] = None,
user: typing.Union[str, int, None] = None,
bucket: typing.Dict = None, **kwargs):
bucket: typing.Optional[dict] = None, **kwargs):
if bucket is None:
bucket = {}
temp_bucket = await self.get_bucket(chat=chat, user=user)

12
aiogram/contrib/fsm_storage/rethinkdb.py
View file

@ -97,7 +97,7 @@ class RethinkDBStorage(BaseStorage):
return await r.table(self._table).get(chat)[user]['state'].default(default or None).run(conn)
async def get_data(self, *, chat: typing.Union[str, int, None] = None, user: typing.Union[str, int, None] = None,
default: typing.Optional[str] = None) -> typing.Dict:
default: typing.Optional[dict] = None) -> dict:
chat, user = map(str, self.check_address(chat=chat, user=user))
async with self.connection() as conn:
return await r.table(self._table).get(chat)[user]['data'].default(default or {}).run(conn)
@ -109,7 +109,7 @@ class RethinkDBStorage(BaseStorage):
await r.table(self._table).insert({'id': chat, user: {'state': state}}, conflict="update").run(conn)
async def set_data(self, *, chat: typing.Union[str, int, None] = None, user: typing.Union[str, int, None] = None,
data: typing.Dict = None):
data: typing.Optional[dict] = None):
chat, user = map(str, self.check_address(chat=chat, user=user))
async with self.connection() as conn:
if await r.table(self._table).get(chat).run(conn):
@ -118,7 +118,7 @@ class RethinkDBStorage(BaseStorage):
await r.table(self._table).insert({'id': chat, user: {'data': data}}).run(conn)
async def update_data(self, *, chat: typing.Union[str, int, None] = None, user: typing.Union[str, int, None] = None,
data: typing.Dict = None,
data: typing.Optional[dict] = None,
**kwargs):
chat, user = map(str, self.check_address(chat=chat, user=user))
async with self.connection() as conn:
@ -128,13 +128,13 @@ class RethinkDBStorage(BaseStorage):
return True
async def get_bucket(self, *, chat: typing.Union[str, int, None] = None, user: typing.Union[str, int, None] = None,
default: typing.Optional[dict] = None) -> typing.Dict:
default: typing.Optional[dict] = None) -> dict:
chat, user = map(str, self.check_address(chat=chat, user=user))
async with self.connection() as conn:
return await r.table(self._table).get(chat)[user]['bucket'].default(default or {}).run(conn)
async def set_bucket(self, *, chat: typing.Union[str, int, None] = None, user: typing.Union[str, int, None] = None,
bucket: typing.Dict = None):
bucket: typing.Optional[dict] = None):
chat, user = map(str, self.check_address(chat=chat, user=user))
async with self.connection() as conn:
if await r.table(self._table).get(chat).run(conn):
@ -143,7 +143,7 @@ class RethinkDBStorage(BaseStorage):
await r.table(self._table).insert({'id': chat, user: {'bucket': bucket}}).run(conn)
async def update_bucket(self, *, chat: typing.Union[str, int, None] = None,
user: typing.Union[str, int, None] = None, bucket: typing.Dict = None,
user: typing.Union[str, int, None] = None, bucket: typing.Optional[dict] = None,
**kwargs):
chat, user = map(str, self.check_address(chat=chat, user=user))
async with self.connection() as conn:

24
aiogram/dispatcher/storage.py
View file

@ -80,7 +80,7 @@ class BaseStorage:
async def get_data(self, *,
chat: typing.Union[str, int, None] = None,
user: typing.Union[str, int, None] = None,
default: typing.Optional[typing.Dict] = None) -> typing.Dict:
default: typing.Optional[typing.Dict] = None) -> dict:
"""
Get state-data for user in chat. Return `default` if no data is provided in storage.
@ -113,7 +113,7 @@ class BaseStorage:
async def set_data(self, *,
chat: typing.Union[str, int, None] = None,
user: typing.Union[str, int, None] = None,
data: typing.Dict = None):
data: typing.Optional[dict] = None):
"""
Set data for user in chat
@ -129,7 +129,7 @@ class BaseStorage:
async def update_data(self, *,
chat: typing.Union[str, int, None] = None,
user: typing.Union[str, int, None] = None,
data: typing.Dict = None,
data: typing.Optional[dict] = None,
**kwargs):
"""
Update data for user in chat
@ -204,7 +204,7 @@ class BaseStorage:
async def get_bucket(self, *,
chat: typing.Union[str, int, None] = None,
user: typing.Union[str, int, None] = None,
default: typing.Optional[dict] = None) -> typing.Dict:
default: typing.Optional[dict] = None) -> dict:
"""
Get bucket for user in chat. Return `default` if no data is provided in storage.
@ -221,7 +221,7 @@ class BaseStorage:
async def set_bucket(self, *,
chat: typing.Union[str, int, None] = None,
user: typing.Union[str, int, None] = None,
bucket: typing.Dict = None):
bucket: typing.Optional[dict] = None):
"""
Set bucket for user in chat
@ -237,7 +237,7 @@ class BaseStorage:
async def update_bucket(self, *,
chat: typing.Union[str, int, None] = None,
user: typing.Union[str, int, None] = None,
bucket: typing.Dict = None,
bucket: typing.Optional[dict] = None,
**kwargs):
"""
Update bucket for user in chat
@ -294,16 +294,16 @@ class FSMContext:
async def get_state(self, default: typing.Optional[str] = None) -> typing.Optional[str]:
return await self.storage.get_state(chat=self.chat, user=self.user, default=self._resolve_state(default))
async def get_data(self, default: typing.Optional[str] = None) -> typing.Dict:
async def get_data(self, default: typing.Optional[dict] = None) -> dict:
return await self.storage.get_data(chat=self.chat, user=self.user, default=default)
async def update_data(self, data: typing.Dict = None, **kwargs):
async def update_data(self, data: typing.Optional[dict] = None, **kwargs):
await self.storage.update_data(chat=self.chat, user=self.user, data=data, **kwargs)
async def set_state(self, state: typing.Union[typing.AnyStr, None] = None):
await self.storage.set_state(chat=self.chat, user=self.user, state=self._resolve_state(state))
async def set_data(self, data: typing.Dict = None):
async def set_data(self, data: typing.Optional[dict] = None):
await self.storage.set_data(chat=self.chat, user=self.user, data=data)
async def reset_state(self, with_data: typing.Optional[bool] = True):
@ -471,14 +471,14 @@ class DisabledStorage(BaseStorage):
async def get_data(self, *,
chat: typing.Union[str, int, None] = None,
user: typing.Union[str, int, None] = None,
default: typing.Optional[str] = None) -> typing.Dict:
default: typing.Optional[str] = None) -> dict:
self._warn()
return {}
async def update_data(self, *,
chat: typing.Union[str, int, None] = None,
user: typing.Union[str, int, None] = None,
data: typing.Dict = None, **kwargs):
data: typing.Optional[dict] = None, **kwargs):
self._warn()
async def set_state(self, *,
@ -490,7 +490,7 @@ class DisabledStorage(BaseStorage):
async def set_data(self, *,
chat: typing.Union[str, int, None] = None,
user: typing.Union[str, int, None] = None,
data: typing.Dict = None):
data: typing.Optional[dict] = None):
self._warn()
@staticmethod

4
aiogram/dispatcher/webhook.py
View file

@ -321,7 +321,7 @@ class BaseResponse:
"""
raise NotImplementedError
def prepare(self) -> typing.Dict:
def prepare(self) -> dict:
"""
You need to override this method.
@ -329,7 +329,7 @@ class BaseResponse:
"""
raise NotImplementedError
def cleanup(self) -> typing.Dict:
def cleanup(self) -> dict:
"""
Cleanup response after preparing. Remove empty fields.