Zergling_man/moment
Zergling_man/moment
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Use typing.TYPE_CHECKING to avoid inner imports

Browse Source
This commit is contained in:
miruka 2020-02-11 16:22:05 -04:00
parent a653a6160a
commit b992db9bfe
9 changed files with 58 additions and 52 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
src/backend/backend.py
View File

@ -13,9 +13,11 @@ from appdirs import AppDirs
from . import __app_name__
from .errors import MatrixError
from .matrix_client import MatrixClient
from .media_cache import MediaCache
from .models import SyncId
from .models.items import Account
from .models.model_store import ModelStore
from .user_files import Accounts, History, Theme, UISettings, UIState
# Logging configuration
log.getLogger().setLevel(log.INFO)
@ -71,7 +73,6 @@ class Backend:
def __init__(self) -> None:
self.appdirs = AppDirs(appname=__app_name__, roaming=True)
from .user_files import Accounts, UISettings, UIState, History
self.saved_accounts: Accounts = Accounts(self)
self.ui_settings: UISettings = UISettings(self)
self.ui_state: UIState = UIState(self)
@ -87,7 +88,6 @@ class Backend:
self.send_locks: DefaultDict[str, asyncio.Lock] = \
DefaultDict(asyncio.Lock) # {room_id: lock}
from .media_cache import MediaCache
cache_dir = Path(self.appdirs.user_cache_dir)
self.media_cache: MediaCache = MediaCache(self, cache_dir)
@ -247,7 +247,6 @@ class Backend:
async def load_settings(self) -> tuple:
"""Return parsed user config files."""
from .user_files import Theme
settings = await self.ui_settings.read()
ui_state = await self.ui_state.read()
history = await self.history.read()

26
src/backend/matrix_client.py
View File

@ -14,19 +14,18 @@ from datetime import datetime
from functools import partial
from pathlib import Path
from typing import (
Any, DefaultDict, Dict, List, NamedTuple, Optional, Set, Tuple, Type,
Union,
TYPE_CHECKING, Any, DefaultDict, Dict, List, NamedTuple, Optional, Set,
Tuple, Type, Union,
)
from urllib.parse import urlparse
from uuid import UUID, uuid4
import cairosvg
from PIL import Image as PILImage
from pymediainfo import MediaInfo
import nio
from nio.crypto import AsyncDataT as UploadData
from nio.crypto import async_generator_from_data
from PIL import Image as PILImage
from pymediainfo import MediaInfo
from . import __app_name__, __display_name__, utils
from .errors import (
@ -34,12 +33,15 @@ from .errors import (
MatrixNotFound, UneededThumbnail,
)
from .html_markdown import HTML_PROCESSOR as HTML
from .models.items import (
Event, Member, Room, TypeSpecifier, Upload, UploadStatus,
)
from .media_cache import Media, Thumbnail
from .models.items import Event, Member, Room, Upload, UploadStatus
from .models.model_store import ModelStore
from .nio_callbacks import NioCallbacks
from .pyotherside_events import AlertRequested
if TYPE_CHECKING:
from .backend import Backend
CryptDict = Dict[str, Any]
@ -103,8 +105,7 @@ class MatrixClient(nio.AsyncClient):
),
)
from .backend import Backend
self.backend: Backend = backend
self.backend: "Backend" = backend
self.models: ModelStore = self.backend.models
self.profile_task: Optional[asyncio.Future] = None
@ -120,7 +121,6 @@ class MatrixClient(nio.AsyncClient):
self.skipped_events: DefaultDict[str, int] = DefaultDict(lambda: 0)
from .nio_callbacks import NioCallbacks
self.nio_callbacks = NioCallbacks(self)
@ -292,11 +292,9 @@ class MatrixClient(nio.AsyncClient):
) -> None:
"""Monitorably upload a file + thumbnail and send the built event."""
# TODO: this function is WAY TOO COMPLEX, and most of it should be
# TODO: this function is way too complex, and most of it should be
# refactored into nio.
from .media_cache import Media, Thumbnail
transaction_id = uuid4()
path = Path(path)
encrypt = room_id in self.encrypted_rooms

13
src/backend/media_cache.py
View File

@ -10,17 +10,18 @@ import shutil
import sys
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any, DefaultDict, Dict, Optional
from typing import TYPE_CHECKING, Any, DefaultDict, Dict, Optional
from urllib.parse import urlparse
import aiofiles
import nio
from PIL import Image as PILImage
import nio
from .backend import Backend
from .utils import Size
if TYPE_CHECKING:
from .backend import Backend
if sys.version_info < (3, 8):
import pyfastcopy # noqa
@ -34,8 +35,8 @@ ACCESS_LOCKS: DefaultDict[str, asyncio.Lock] = DefaultDict(asyncio.Lock)
class MediaCache:
"""Matrix downloaded media cache."""
backend: Backend = field()
base_dir: Path = field()
backend: "Backend" = field()
base_dir: Path = field()
def __post_init__(self) -> None:

16
src/backend/models/model.py
View File

@ -1,13 +1,15 @@
# SPDX-License-Identifier: LGPL-3.0-or-later
from threading import Lock
from typing import Any, Dict, Iterator, List, MutableMapping
from typing import TYPE_CHECKING, Any, Dict, Iterator, List, MutableMapping
from ..pyotherside_events import (
ModelCleared, ModelItemDeleted, ModelItemInserted,
)
from . import SyncId
from .model_item import ModelItem
if TYPE_CHECKING:
from .model_item import ModelItem
class Model(MutableMapping):
@ -30,10 +32,10 @@ class Model(MutableMapping):
"""
def __init__(self, sync_id: SyncId) -> None:
self.sync_id: SyncId = sync_id
self._data: Dict[Any, ModelItem] = {}
self._sorted_data: List[ModelItem] = []
self._write_lock: Lock = Lock()
self.sync_id: SyncId = sync_id
self._data: Dict[Any, "ModelItem"] = {}
self._sorted_data: List["ModelItem"] = []
self._write_lock: Lock = Lock()
def __repr__(self) -> str:
@ -58,7 +60,7 @@ class Model(MutableMapping):
return self._data[key]
def __setitem__(self, key, value: ModelItem) -> None:
def __setitem__(self, key, value: "ModelItem") -> None:
"""Merge new item with an existing one if possible, else add it.
If an existing item with the passed `key` is found, its fields will be

12
src/backend/models/model_item.py
View File

@ -1,6 +1,12 @@
# SPDX-License-Identifier: LGPL-3.0-or-later
from typing import Any, Dict, Optional
from typing import TYPE_CHECKING, Any, Dict, Optional
from ..pyotherside_events import ModelItemFieldChanged
from ..utils import serialize_value_for_qml
if TYPE_CHECKING:
from .model import Model
class ModelItem:
@ -18,7 +24,6 @@ class ModelItem:
"""
def __new__(cls, *_args, **_kwargs) -> "ModelItem":
from .model import Model
cls.parent_model: Optional[Model] = None
return super().__new__(cls)
@ -39,7 +44,6 @@ class ModelItem:
self.parent_model._sorted_data.sort()
new_index = self.parent_model._sorted_data.index(self)
from ..pyotherside_events import ModelItemFieldChanged
ModelItemFieldChanged(
self.parent_model.sync_id, old_index, new_index, name, value,
)
@ -53,8 +57,6 @@ class ModelItem:
def serialized(self) -> Dict[str, Any]:
"""Return this item as a dict ready to be passed to QML."""
from ..utils import serialize_value_for_qml
return {
name: serialize_value_for_qml(getattr(self, name), json_lists=True)
for name in dir(self)

8
src/backend/nio_callbacks.py
View File

@ -6,16 +6,18 @@ import logging as log
from contextlib import suppress
from dataclasses import dataclass, field
from datetime import datetime
from typing import Optional, Tuple
from typing import TYPE_CHECKING, Optional, Tuple
from urllib.parse import quote
import nio
from . import utils
from .html_markdown import HTML_PROCESSOR
from .matrix_client import MatrixClient
from .models.items import TypeSpecifier
if TYPE_CHECKING:
from .matrix_client import MatrixClient
@dataclass
class NioCallbacks:
@ -30,7 +32,7 @@ class NioCallbacks:
These are processed from QML, to allow translations of the strings.
"""
client: MatrixClient = field()
client: "MatrixClient" = field()
def __post_init__(self) -> None:

12
src/backend/pyotherside_events.py
View File

@ -2,14 +2,16 @@
from abc import ABC
from dataclasses import dataclass, field
from typing import Any, Optional
from typing import TYPE_CHECKING, Any, Optional
import pyotherside
from .models import SyncId
from .models.model_item import ModelItem
from .utils import serialize_value_for_qml
if TYPE_CHECKING:
from .models.model_item import ModelItem
@dataclass
class PyOtherSideEvent:
@ -69,9 +71,9 @@ class ModelEvent(ABC, PyOtherSideEvent):
@dataclass
class ModelItemInserted(ModelEvent):
sync_id: SyncId = field()
index: int = field()
item: ModelItem = field()
sync_id: SyncId = field()
index: int = field()
item: "ModelItem" = field()
@dataclass

10
src/backend/user_files.py
View File

@ -7,14 +7,16 @@ import json
import logging as log
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any, ClassVar, Dict, Optional
from typing import TYPE_CHECKING, Any, ClassVar, Dict, Optional
import aiofiles
from .backend import Backend
from .theme_parser import convert_to_qml
from .utils import dict_update_recursive
if TYPE_CHECKING:
from .backend import Backend
JsonData = Dict[str, Any]
WRITE_LOCK = asyncio.Lock()
@ -26,8 +28,8 @@ class DataFile:
is_config: ClassVar[bool] = False
backend: Backend = field(repr=False)
filename: str = field()
backend: "Backend" = field(repr=False)
filename: str = field()
_to_write: Optional[str] = field(init=False, default=None)

8
src/backend/utils.py
View File

@ -21,8 +21,6 @@ from aiofiles.threadpool.binary import AsyncBufferedReader
from nio.crypto import AsyncDataT as File
from nio.crypto import async_generator_from_data
from .models.model_item import ModelItem
Size = Tuple[int, int]
auto = autostr
@ -143,12 +141,12 @@ def serialize_value_for_qml(value: Any, json_lists: bool = False) -> Any:
if json_lists and isinstance(value, list):
return json.dumps(value)
if hasattr(value, "serialized"):
return value.serialized
if hasattr(value, "__class__") and issubclass(value.__class__, Enum):
return value.value
if isinstance(value, ModelItem):
return value.serialized
if isinstance(value, Path):
return f"file://{value!s}"