style: prefer PEP 604/585 type hints
This commit is contained in:
@@ -1,6 +1,6 @@
|
||||
"""Command implementations for Skywipe"""
|
||||
|
||||
from typing import Callable, Dict, Optional, Any
|
||||
from typing import Callable, Any
|
||||
from .configure import Configuration
|
||||
from .operations import Operation
|
||||
from .post_analysis import PostAnalyzer
|
||||
@@ -98,16 +98,16 @@ class CommandRegistry:
|
||||
self._help_texts[name] = help_text
|
||||
self._requires_config[name] = requires_config
|
||||
|
||||
def get_handler(self, name: str) -> Optional[CommandHandler]:
|
||||
def get_handler(self, name: str) -> CommandHandler | None:
|
||||
return self._commands.get(name)
|
||||
|
||||
def get_help_text(self, name: str) -> Optional[str]:
|
||||
def get_help_text(self, name: str) -> str | None:
|
||||
return self._help_texts.get(name)
|
||||
|
||||
def requires_config(self, name: str) -> bool:
|
||||
return self._requires_config.get(name, True)
|
||||
|
||||
def get_all_commands(self) -> Dict[str, str]:
|
||||
def get_all_commands(self) -> dict[str, str]:
|
||||
return self._help_texts.copy()
|
||||
|
||||
def execute(self, name: str, skip_confirmation: bool = False):
|
||||
@@ -128,8 +128,8 @@ def _create_operation_handler(
|
||||
confirmation_message: str,
|
||||
operation_name: str,
|
||||
strategy_type: str = "feed",
|
||||
collection: Optional[str] = None,
|
||||
filter_fn: Optional[Callable[[Any], bool]] = None
|
||||
collection: str | None = None,
|
||||
filter_fn: Callable[[Any], bool] | None = None
|
||||
) -> CommandHandler:
|
||||
logger = get_logger()
|
||||
|
||||
|
||||
@@ -3,7 +3,6 @@
|
||||
import logging
|
||||
import sys
|
||||
from pathlib import Path
|
||||
from typing import Optional
|
||||
|
||||
|
||||
class ProgressTracker:
|
||||
@@ -14,7 +13,7 @@ class ProgressTracker:
|
||||
def update(self, count: int = 1):
|
||||
self.current += count
|
||||
|
||||
def batch(self, batch_num: int, batch_size: int, total_batches: Optional[int] = None):
|
||||
def batch(self, batch_num: int, batch_size: int, total_batches: int | None = None):
|
||||
logger = logging.getLogger("skywipe.progress")
|
||||
if total_batches:
|
||||
logger.info(
|
||||
@@ -35,7 +34,7 @@ class LevelFilter(logging.Filter):
|
||||
return self.min_level <= record.levelno <= self.max_level
|
||||
|
||||
|
||||
def setup_logger(verbose: bool = False, log_file: Optional[Path] = None) -> logging.Logger:
|
||||
def setup_logger(verbose: bool = False, log_file: Path | None = None) -> logging.Logger:
|
||||
logger = logging.getLogger("skywipe")
|
||||
target_level = logging.DEBUG if verbose else logging.INFO
|
||||
logger.setLevel(target_level)
|
||||
|
||||
@@ -1,7 +1,7 @@
|
||||
"""Shared operation utilities and strategies for Skywipe"""
|
||||
|
||||
import time
|
||||
from typing import Callable, Optional, Any
|
||||
from typing import Callable, Any
|
||||
from atproto import models
|
||||
|
||||
from .auth import Auth
|
||||
@@ -60,7 +60,7 @@ class RecordDeletionStrategy(BaseStrategy):
|
||||
def __init__(self, collection: str):
|
||||
self.collection = collection
|
||||
|
||||
def fetch(self, context: OperationContext, cursor: Optional[str] = None):
|
||||
def fetch(self, context: OperationContext, cursor: str | None = None):
|
||||
list_params = models.ComAtprotoRepoListRecords.Params(
|
||||
repo=context.did,
|
||||
collection=self.collection,
|
||||
@@ -85,7 +85,7 @@ class RecordDeletionStrategy(BaseStrategy):
|
||||
|
||||
|
||||
class FeedStrategy(BaseStrategy):
|
||||
def fetch(self, context: OperationContext, cursor: Optional[str] = None):
|
||||
def fetch(self, context: OperationContext, cursor: str | None = None):
|
||||
if cursor:
|
||||
return context.client.get_author_feed(
|
||||
actor=context.did, limit=context.batch_size, cursor=cursor
|
||||
@@ -102,7 +102,7 @@ class FeedStrategy(BaseStrategy):
|
||||
|
||||
|
||||
class BookmarkStrategy(BaseStrategy):
|
||||
def fetch(self, context: OperationContext, cursor: Optional[str] = None):
|
||||
def fetch(self, context: OperationContext, cursor: str | None = None):
|
||||
get_params = models.AppBskyBookmarkGetBookmarks.Params(
|
||||
limit=context.batch_size,
|
||||
cursor=cursor
|
||||
@@ -139,8 +139,8 @@ class Operation:
|
||||
self,
|
||||
operation_name: str,
|
||||
strategy_type: str = "feed",
|
||||
collection: Optional[str] = None,
|
||||
filter_fn: Optional[Callable[[Any], bool]] = None,
|
||||
collection: str | None = None,
|
||||
filter_fn: Callable[[Any], bool] | None = None,
|
||||
client=None,
|
||||
config_data=None
|
||||
):
|
||||
|
||||
@@ -2,14 +2,17 @@
|
||||
|
||||
import sys
|
||||
import logging
|
||||
from typing import Optional
|
||||
from .logger import get_logger
|
||||
|
||||
|
||||
CONFIRM_RESPONSES = {"yes", "y"}
|
||||
|
||||
|
||||
def require_confirmation(operation: str, skip_confirmation: bool = False, logger: Optional[logging.Logger] = None) -> None:
|
||||
def require_confirmation(
|
||||
operation: str,
|
||||
skip_confirmation: bool = False,
|
||||
logger: logging.Logger | None = None
|
||||
) -> None:
|
||||
if skip_confirmation:
|
||||
return
|
||||
|
||||
|
||||
Reference in New Issue
Block a user