Skip to content

Commit

Permalink
Add more typing (#464)
Browse files Browse the repository at this point in the history
  • Loading branch information
sobolevn authored Jun 28, 2024
1 parent 19ad3b3 commit 3543a6b
Show file tree
Hide file tree
Showing 25 changed files with 367 additions and 775 deletions.
13 changes: 11 additions & 2 deletions CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -2,11 +2,20 @@

We follow Semantic Versions since the `0.1.0` release.

## Version

## Version 1.4.0

### Features

- Adds Python 3.12 support
- Drops Python 3.8 support
- Updates `typing_extensions` to `>=4,<5`
- Adds more typing to the project

### Fixes

- Fix getting the `statement_timeout` setting name on MariaDB servers
- Fixes getting the `statement_timeout` setting name on MariaDB servers
- Fixes delayed apps cache


## Version 1.3.0
Expand Down
2 changes: 1 addition & 1 deletion README.md
Original file line number Diff line number Diff line change
Expand Up @@ -201,7 +201,7 @@ That's how it can be used:
```python
import pytest

@pytest.mark.django_db()
@pytest.mark.django_db
def test_pytest_plugin_initial(migrator):
"""Ensures that the initial migration works."""
old_state = migrator.apply_initial_migration(('main_app', None))
Expand Down
11 changes: 8 additions & 3 deletions django_test_migrations/checks/autonames.py
Original file line number Diff line number Diff line change
@@ -1,5 +1,5 @@
from fnmatch import fnmatch
from typing import FrozenSet, List, Tuple
from typing import FrozenSet, Sequence, Tuple

from django.conf import settings
from django.core.checks import CheckMessage, Warning
Expand Down Expand Up @@ -47,7 +47,10 @@ def _build_ignores() -> _IgnoreSpec:
return ignored_apps, ignored_migrations


def check_migration_names(*args, **kwargs) -> List[CheckMessage]:
def check_migration_names(
*args: object,
**kwargs: object,
) -> Sequence[CheckMessage]:
"""
Finds automatic names in available migrations.
Expand Down Expand Up @@ -84,4 +87,6 @@ def check_migration_names(*args, **kwargs) -> List[CheckMessage]:
return messages


CHECKS: Final = (check_migration_names,)
CHECKS: Final = (
check_migration_names,
)
14 changes: 10 additions & 4 deletions django_test_migrations/contrib/django_checks.py
Original file line number Diff line number Diff line change
Expand Up @@ -40,10 +40,13 @@ class AutoNames(AppConfig):
#: Part of Django API.
name = autonames.CHECK_NAME

def ready(self):
def ready(self) -> None:
"""That's how we register our check when apps are ready."""
for check in autonames.CHECKS:
checks.register(check, checks.Tags.compatibility)
checks.register( # type: ignore[call-overload]
check,
checks.Tags.compatibility,
)


@final
Expand All @@ -67,7 +70,10 @@ class DatabaseConfiguration(AppConfig):
#: Part of Django API.
name = database_configuration.CHECK_NAME

def ready(self):
def ready(self) -> None:
"""Register database configuration checks."""
for check in database_configuration.CHECKS:
checks.register(check, checks.Tags.database)
checks.register( # type: ignore[call-overload]
check,
checks.Tags.database,
)
40 changes: 29 additions & 11 deletions django_test_migrations/contrib/pytest_plugin.py
Original file line number Diff line number Diff line change
@@ -1,12 +1,15 @@
from typing import Optional
from typing import TYPE_CHECKING, List, Optional, Protocol

import pytest
from django.db import DEFAULT_DB_ALIAS

from django_test_migrations.constants import MIGRATION_TEST_MARKER

if TYPE_CHECKING:
from django_test_migrations.migrator import Migrator

def pytest_load_initial_conftests(early_config):

def pytest_load_initial_conftests(early_config: pytest.Config) -> None:
"""Register pytest's markers."""
early_config.addinivalue_line(
'markers',
Expand All @@ -16,8 +19,12 @@ def pytest_load_initial_conftests(early_config):
)


def pytest_collection_modifyitems(session, items): # noqa: WPS110
"""Mark all tests using ``migrator_factory`` fixture with proper marks.
def pytest_collection_modifyitems(
session: pytest.Session,
items: List[pytest.Item], # noqa: WPS110
) -> None:
"""
Mark all tests using ``migrator_factory`` fixture with proper marks.
Add ``MIGRATION_TEST_MARKER`` marker to all items using
``migrator_factory`` fixture.
Expand All @@ -28,16 +35,27 @@ def pytest_collection_modifyitems(session, items): # noqa: WPS110
pytest_item.add_marker(MIGRATION_TEST_MARKER)


@pytest.fixture()
def migrator_factory(request, transactional_db, django_db_use_migrations):
class MigratorFactory(Protocol):
"""Protocol for `migrator_factory` fixture."""

def __call__(self, database_name: Optional[str] = None) -> 'Migrator':
"""It only has a `__call__` magic method."""


@pytest.fixture
def migrator_factory(
request: pytest.FixtureRequest,
transactional_db: None,
django_db_use_migrations: bool,
) -> MigratorFactory:
"""
Pytest fixture to create migrators inside the pytest tests.
How? Here's an example.
.. code:: python
@pytest.mark.django_db()
@pytest.mark.django_db
def test_migration(migrator_factory):
migrator = migrator_factory('custom_db_alias')
old_state = migrator.apply_initial_migration(('main_app', None))
Expand All @@ -56,7 +74,7 @@ def test_migration(migrator_factory):
That's why we cannot import ``Migrator`` on a module level.
Because it won't be caught be coverage later on.
"""
from django_test_migrations.migrator import Migrator # noqa: WPS433
from django_test_migrations.migrator import Migrator # noqa: WPS433, WPS474

if not django_db_use_migrations:
pytest.skip('--nomigrations was specified')
Expand All @@ -68,8 +86,8 @@ def factory(database_name: Optional[str] = None) -> Migrator:
return factory


@pytest.fixture()
def migrator(migrator_factory): # noqa: WPS442
@pytest.fixture
def migrator(migrator_factory: MigratorFactory) -> 'Migrator':
"""
Useful alias for ``'default'`` database in ``django``.
Expand All @@ -79,7 +97,7 @@ def migrator(migrator_factory): # noqa: WPS442
.. code:: python
@pytest.mark.django_db()
@pytest.mark.django_db
def test_migration(migrator):
old_state = migrator.apply_initial_migration(('main_app', None))
new_state = migrator.apply_tested_migration(
Expand Down
4 changes: 2 additions & 2 deletions django_test_migrations/contrib/unittest_case.py
Original file line number Diff line number Diff line change
Expand Up @@ -54,11 +54,11 @@ def tearDown(self) -> None:
self._migrator.reset()
super().tearDown()

def _pre_setup(self):
def _pre_setup(self) -> None:
self._pre_migrate_receivers, pre_migrate.receivers = ( # noqa: WPS414
pre_migrate.receivers, [],
)
self._post_migrate_receivers, post_migrate.receivers = ( # noqa: WPS414
post_migrate.receivers, [],
)
super()._pre_setup()
super()._pre_setup() # type: ignore[misc]
2 changes: 1 addition & 1 deletion django_test_migrations/db/backends/base/configuration.py
Original file line number Diff line number Diff line change
Expand Up @@ -17,7 +17,7 @@ class BaseDatabaseConfiguration(abc.ABC):
vendor: ClassVar[str]

@classmethod
def __init_subclass__(cls, **kwargs) -> None:
def __init_subclass__(cls, **kwargs: object) -> None:
"""Register ``BaseDatabaseConfiguration`` subclass of db ``vendor``."""
if not inspect.isabstract(cls):
database_configuration_registry.setdefault(cls.vendor, cls)
Expand Down
3 changes: 2 additions & 1 deletion django_test_migrations/db/backends/mysql/configuration.py
Original file line number Diff line number Diff line change
@@ -1,4 +1,5 @@
from functools import cached_property
from typing import cast

from typing_extensions import final

Expand All @@ -23,7 +24,7 @@ def get_setting_value(self, name: str) -> DatabaseSettingValue:
setting_value = cursor.fetchone()
if not setting_value:
return super().get_setting_value(name)
return setting_value[0]
return cast(DatabaseSettingValue, setting_value[0])

@cached_property
def version(self) -> str:
Expand Down
Original file line number Diff line number Diff line change
@@ -1,3 +1,5 @@
from typing import cast

from typing_extensions import final

from django_test_migrations.db.backends.base.configuration import (
Expand Down Expand Up @@ -26,4 +28,4 @@ def get_setting_value(self, name: str) -> DatabaseSettingValue:
setting_value = cursor.fetchone()
if not setting_value:
return super().get_setting_value(name)
return setting_value[0]
return cast(DatabaseSettingValue, setting_value[0])
5 changes: 4 additions & 1 deletion django_test_migrations/db/checks/statement_timeout.py
Original file line number Diff line number Diff line change
Expand Up @@ -17,7 +17,10 @@
STATEMENT_TIMEOUT_MINUTES_UPPER_LIMIT: Final = 30


def check_statement_timeout_setting(*args, **kwargs) -> List[CheckMessage]:
def check_statement_timeout_setting(
*args: object,
**kwargs: object,
) -> List[CheckMessage]:
"""Check if statements' timeout settings is properly configured."""
messages: List[CheckMessage] = []
for connection in connections.all():
Expand Down
18 changes: 11 additions & 7 deletions django_test_migrations/migrator.py
Original file line number Diff line number Diff line change
@@ -1,4 +1,4 @@
from typing import Optional
from typing import List, Optional

from django.core.management import call_command
from django.core.management.color import no_style
Expand All @@ -10,10 +10,14 @@
from django_test_migrations.logic.migrations import normalize
from django_test_migrations.plan import truncate_plan
from django_test_migrations.signals import mute_migrate_signals
from django_test_migrations.types import MigrationPlan, MigrationSpec
from django_test_migrations.types import (
MigrationPlan,
MigrationSpec,
MigrationTarget,
)


class Migrator(object):
class Migrator:
"""
Class to manage your migrations and app state.
Expand All @@ -40,7 +44,7 @@ def __init__(

def apply_initial_migration(self, targets: MigrationSpec) -> ProjectState:
"""Reverse back to the original migration."""
targets = normalize(targets)
migration_targets = normalize(targets)

style = no_style()
# start from clean database state
Expand All @@ -53,12 +57,12 @@ def apply_initial_migration(self, targets: MigrationSpec) -> ProjectState:
self._executor.loader.graph.leaf_nodes(),
clean_start=True,
)
plan = truncate_plan(targets, full_plan)
plan = truncate_plan(migration_targets, full_plan)

# apply all migrations from generated plan on clean database
# (only forward, so any unexpected migration won't be applied)
# to restore database state before tested migration
return self._migrate(targets, plan=plan)
return self._migrate(migration_targets, plan=plan)

def apply_tested_migration(self, targets: MigrationSpec) -> ProjectState:
"""Apply the next migration."""
Expand All @@ -77,7 +81,7 @@ def reset(self) -> None:

def _migrate(
self,
migration_targets: MigrationSpec,
migration_targets: List[MigrationTarget],
plan: Optional[MigrationPlan] = None,
) -> ProjectState:
with mute_migrate_signals():
Expand Down
5 changes: 4 additions & 1 deletion django_test_migrations/signals.py
Original file line number Diff line number Diff line change
@@ -1,10 +1,13 @@
from __future__ import annotations

from contextlib import contextmanager
from typing import Any, Iterator

from django.db.models.signals import post_migrate, pre_migrate


@contextmanager
def mute_migrate_signals():
def mute_migrate_signals() -> Iterator[tuple[Any, Any]]:
"""Context manager to mute migration-related signals."""
pre_migrate_receivers = pre_migrate.receivers
post_migrate_receivers = post_migrate.receivers
Expand Down
11 changes: 6 additions & 5 deletions django_test_migrations/types.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,15 +3,16 @@
from django.db.backends.base.base import BaseDatabaseWrapper
from django.db.migrations import Migration
from django.utils.connection import ConnectionProxy
from typing_extensions import TypeAlias

# Migration target: (app_name, migration_name)
# Regular or rollback migration: 0001 -> 0002, or 0002 -> 0001
# Rollback migration to initial state: 0001 -> None
MigrationTarget = Tuple[str, Optional[str]]
MigrationSpec = Union[MigrationTarget, List[MigrationTarget]]
MigrationTarget: TypeAlias = Tuple[str, Optional[str]]
MigrationSpec: TypeAlias = Union[MigrationTarget, List[MigrationTarget]]

MigrationPlan = List[Tuple[Migration, bool]]
MigrationPlan: TypeAlias = List[Tuple[Migration, bool]]

AnyConnection = Union[ConnectionProxy, BaseDatabaseWrapper]
AnyConnection: TypeAlias = Union[ConnectionProxy, BaseDatabaseWrapper]

DatabaseSettingValue = Union[str, int]
DatabaseSettingValue: TypeAlias = Union[str, int]
Loading

0 comments on commit 3543a6b

Please sign in to comment.