Kharec/skywipe
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor: add basestrategy and make all class inherits to get rid of get_cursor()

Browse Source
This commit is contained in:
Sandro CURY CAZZANIGA
2025-12-20 20:56:22 +01:00
parent 97dd55981b
commit c2aab71955
1 changed files with 8 additions and 12 deletions
Download Patch File Download Diff File Expand all files Collapse all files

20
skywipe/operations.py
View File

@@ -21,7 +21,12 @@ class OperationContext:
self.did = self.client.me.did self.did = self.client.me.did
class RecordDeletionStrategy: class BaseStrategy:
def get_cursor(self, response):
return response.cursor
class RecordDeletionStrategy(BaseStrategy):
def __init__(self, collection: str): def __init__(self, collection: str):
self.collection = collection self.collection = collection
@@ -37,9 +42,6 @@ class RecordDeletionStrategy:
def extract_items(self, response): def extract_items(self, response):
return response.records return response.records
def get_cursor(self, response):
return response.cursor
def process_item(self, record, context: OperationContext): def process_item(self, record, context: OperationContext):
record_uri = record.uri record_uri = record.uri
rkey = record_uri.rsplit("/", 1)[-1] rkey = record_uri.rsplit("/", 1)[-1]
@@ -52,7 +54,7 @@ class RecordDeletionStrategy:
context.logger.debug(f"Deleted: {record_uri}") context.logger.debug(f"Deleted: {record_uri}")
class FeedStrategy: class FeedStrategy(BaseStrategy):
def fetch(self, context: OperationContext, cursor: Optional[str] = None): def fetch(self, context: OperationContext, cursor: Optional[str] = None):
if cursor: if cursor:
return context.client.get_author_feed( return context.client.get_author_feed(
@@ -63,16 +65,13 @@ class FeedStrategy:
def extract_items(self, response): def extract_items(self, response):
return response.feed return response.feed
def get_cursor(self, response):
return response.cursor
def process_item(self, post, context: OperationContext): def process_item(self, post, context: OperationContext):
uri = post.post.uri uri = post.post.uri
context.client.delete_post(uri) context.client.delete_post(uri)
context.logger.debug(f"Deleted post: {uri}") context.logger.debug(f"Deleted post: {uri}")
class BookmarkStrategy: class BookmarkStrategy(BaseStrategy):
def fetch(self, context: OperationContext, cursor: Optional[str] = None): def fetch(self, context: OperationContext, cursor: Optional[str] = None):
get_params = models.AppBskyBookmarkGetBookmarks.Params( get_params = models.AppBskyBookmarkGetBookmarks.Params(
limit=context.batch_size, limit=context.batch_size,
@@ -83,9 +82,6 @@ class BookmarkStrategy:
def extract_items(self, response): def extract_items(self, response):
return response.bookmarks return response.bookmarks
def get_cursor(self, response):
return response.cursor
def process_item(self, bookmark, context: OperationContext): def process_item(self, bookmark, context: OperationContext):
bookmark_uri = self._extract_bookmark_uri(bookmark) bookmark_uri = self._extract_bookmark_uri(bookmark)
if not bookmark_uri: if not bookmark_uri: