fix formatting

This commit is contained in:
abhi1992002
2025-01-03 12:35:03 +05:30
parent eb25e731fc
commit fb9d42f466
11 changed files with 93 additions and 61 deletions

View File

@@ -79,6 +79,7 @@ class TwitterUnfollowListBlock(Block):
except Exception as e:
yield "error", handle_tweepy_exception(e)
class TwitterFollowListBlock(Block):
"""
Follows a Twitter list for the authenticated user

View File

@@ -283,7 +283,7 @@ class TwitterGetOwnedListsBlock(Block):
response = cast(Response, client.get_owned_lists(**params))
meta = {}
included={}
included = {}
list_ids = []
list_names = []
next_token = None
@@ -297,8 +297,12 @@ class TwitterGetOwnedListsBlock(Block):
if response.data:
data = ResponseDataSerializer.serialize_list(response.data)
list_ids = [str(item.id) for item in response.data if hasattr(item, 'id')]
list_names = [item.name for item in response.data if hasattr(item, 'name')]
list_ids = [
str(item.id) for item in response.data if hasattr(item, "id")
]
list_names = [
item.name for item in response.data if hasattr(item, "name")
]
return data, included, meta, list_ids, list_names, next_token

View File

@@ -480,7 +480,7 @@ class TwitterGetListMembershipsBlock(Block):
next_token = meta.get("next_token")
if response.includes:
included = IncludesSerializer.serialize(response.includes)
included = IncludesSerializer.serialize(response.includes)
if response.data:
data = ResponseDataSerializer.serialize_list(response.data)

View File

@@ -77,6 +77,7 @@ class TwitterDeleteListBlock(Block):
except Exception as e:
yield "error", handle_tweepy_exception(e)
class TwitterUpdateListBlock(Block):
"""
Updates a Twitter List owned by the authenticated user
@@ -132,7 +133,10 @@ class TwitterUpdateListBlock(Block):
@staticmethod
def update_list(
credentials: TwitterCredentials, list_id: str, name: str | None, description: str | None
credentials: TwitterCredentials,
list_id: str,
name: str | None,
description: str | None,
):
try:
client = tweepy.Client(
@@ -161,16 +165,14 @@ class TwitterUpdateListBlock(Block):
) -> BlockOutput:
try:
success = self.update_list(
credentials,
input_data.list_id,
input_data.name,
input_data.description
credentials, input_data.list_id, input_data.name, input_data.description
)
yield "success", success
except Exception as e:
yield "error", handle_tweepy_exception(e)
class TwitterCreateListBlock(Block):
"""
Creates a Twitter List owned by the authenticated user
@@ -229,7 +231,10 @@ class TwitterCreateListBlock(Block):
@staticmethod
def create_list(
credentials: TwitterCredentials, name: str, description: str | None, private: bool
credentials: TwitterCredentials,
name: str,
description: str | None,
private: bool,
):
try:
client = tweepy.Client(

View File

@@ -244,7 +244,6 @@ class TwitterGetPinnedListsBlock(Block):
if response.includes:
included = IncludesSerializer.serialize(response.includes)
if response.data:
data = ResponseDataSerializer.serialize_list(response.data)
list_ids = [str(item.id) for item in response.data]

View File

@@ -1,7 +1,7 @@
from typing import Literal, Union, cast
from pydantic import BaseModel
import tweepy
from pydantic import BaseModel
from tweepy.client import Response
from backend.blocks.twitter._auth import (
@@ -38,8 +38,9 @@ from backend.blocks.twitter.tweepy_exceptions import handle_tweepy_exception
from backend.data.block import Block, BlockCategory, BlockOutput, BlockSchema
from backend.data.model import SchemaField
class SpaceList(BaseModel):
discriminator: Literal['space_list']
discriminator: Literal["space_list"]
space_ids: list[str] = SchemaField(
description="List of Space IDs to lookup (up to 100)",
placeholder="Enter Space IDs",
@@ -47,8 +48,9 @@ class SpaceList(BaseModel):
advanced=False,
)
class UserList(BaseModel):
discriminator: Literal['user_list']
discriminator: Literal["user_list"]
user_ids: list[str] = SchemaField(
description="List of user IDs to lookup their Spaces (up to 100)",
placeholder="Enter user IDs",
@@ -56,6 +58,7 @@ class UserList(BaseModel):
advanced=False,
)
class TwitterGetSpacesBlock(Block):
"""
Gets information about multiple Twitter Spaces specified by Space IDs or creator user IDs
@@ -67,7 +70,7 @@ class TwitterGetSpacesBlock(Block):
)
identifier: Union[SpaceList, UserList] = SchemaField(
discriminator='discriminator',
discriminator="discriminator",
description="Choose whether to lookup spaces by their IDs or by creator user IDs",
advanced=False,
)
@@ -94,7 +97,7 @@ class TwitterGetSpacesBlock(Block):
test_input={
"identifier": {
"discriminator": "space_list",
"space_ids": ["1DXxyRYNejbKM"]
"space_ids": ["1DXxyRYNejbKM"],
},
"credentials": TEST_CREDENTIALS_INPUT,
"expansions": None,
@@ -146,8 +149,12 @@ class TwitterGetSpacesBlock(Block):
)
params = {
"ids": identifier.space_ids if isinstance(identifier, SpaceList) else None,
"user_ids": identifier.user_ids if isinstance(identifier, UserList) else None,
"ids": (
identifier.space_ids if isinstance(identifier, SpaceList) else None
),
"user_ids": (
identifier.user_ids if isinstance(identifier, UserList) else None
),
}
params = (
@@ -206,6 +213,7 @@ class TwitterGetSpacesBlock(Block):
except Exception as e:
yield "error", handle_tweepy_exception(e)
class TwitterGetSpaceByIdBlock(Block):
"""
Gets information about a single Twitter Space specified by Space ID
@@ -366,6 +374,7 @@ class TwitterGetSpaceByIdBlock(Block):
except Exception as e:
yield "error", handle_tweepy_exception(e)
# Not tested yet, might have some problem
class TwitterGetSpaceBuyersBlock(Block):
"""

View File

@@ -1,8 +1,8 @@
from datetime import datetime
from typing import List, Literal, Optional, Union, cast
from pydantic import BaseModel
import tweepy
from pydantic import BaseModel
from tweepy.client import Response
from backend.blocks.twitter._auth import (
@@ -37,28 +37,34 @@ from backend.blocks.twitter.tweepy_exceptions import handle_tweepy_exception
from backend.data.block import Block, BlockCategory, BlockOutput, BlockSchema
from backend.data.model import SchemaField
class Media(BaseModel):
discriminator: Literal['media']
discriminator: Literal["media"]
media_ids: Optional[List[str]] = None
media_tagged_user_ids: Optional[List[str]] = None
class DeepLink(BaseModel):
discriminator: Literal['deep_link']
discriminator: Literal["deep_link"]
direct_message_deep_link: Optional[str] = None
class Poll(BaseModel):
discriminator: Literal['poll']
discriminator: Literal["poll"]
poll_options: Optional[List[str]] = None
poll_duration_minutes: Optional[int] = None
class Place(BaseModel):
discriminator: Literal['place']
discriminator: Literal["place"]
place_id: Optional[str] = None
class Quote(BaseModel):
discriminator: Literal['quote']
discriminator: Literal["quote"]
quote_tweet_id: Optional[str] = None
class TwitterPostTweetBlock(Block):
"""
Create a tweet on Twitter with the option to include one additional element such as a media, quote, or deep link.
@@ -72,8 +78,8 @@ class TwitterPostTweetBlock(Block):
tweet_text: str | None = SchemaField(
description="Text of the tweet to post",
placeholder="Enter your tweet",
default = None,
advanced= False
default=None,
advanced=False,
)
for_super_followers_only: bool = SchemaField(
@@ -83,8 +89,8 @@ class TwitterPostTweetBlock(Block):
default=False,
)
attachment: Union[Media, DeepLink, Poll, Place, Quote] | None = SchemaField(
discriminator='discriminator',
attachment: Union[Media, DeepLink, Poll, Place, Quote] | None = SchemaField(
discriminator="discriminator",
description="Additional tweet data (media, deep link, poll, place or quote)",
advanced=True,
)
@@ -129,7 +135,7 @@ class TwitterPostTweetBlock(Block):
"credentials": TEST_CREDENTIALS_INPUT,
"attachment": {
"discriminator": "deep_link",
"direct_message_deep_link": "https://twitter.com/messages/compose"
"direct_message_deep_link": "https://twitter.com/messages/compose",
},
"for_super_followers_only": False,
"exclude_reply_user_ids": [],
@@ -170,12 +176,14 @@ class TwitterPostTweetBlock(Block):
.add_reply_settings(
exclude_reply_user_ids or [],
in_reply_to_tweet_id or "",
reply_settings
reply_settings,
)
)
if isinstance(attachment, Media):
params.add_media(attachment.media_ids or [], attachment.media_tagged_user_ids or [])
params.add_media(
attachment.media_ids or [], attachment.media_tagged_user_ids or []
)
elif isinstance(attachment, DeepLink):
params.add_deep_link(attachment.direct_message_deep_link or "")
elif isinstance(attachment, Poll):

View File

@@ -52,9 +52,7 @@ class TwitterGetQuoteTweetsBlock(Block):
)
exclude: TweetExcludesFilter | None = SchemaField(
description="Types of tweets to exclude",
advanced=True,
default=None
description="Types of tweets to exclude", advanced=True, default=None
)
pagination_token: str | None = SchemaField(

View File

@@ -279,6 +279,7 @@ class TwitterGetUserMentionsBlock(Block):
except Exception as e:
yield "error", handle_tweepy_exception(e)
class TwitterGetHomeTimelineBlock(Block):
"""
Returns a collection of the most recent Tweets and Retweets posted by you and users you follow
@@ -509,6 +510,7 @@ class TwitterGetHomeTimelineBlock(Block):
except Exception as e:
yield "error", handle_tweepy_exception(e)
class TwitterGetUserTweetsBlock(Block):
"""
Returns Tweets composed by a single user, specified by the requested user ID

View File

@@ -1,7 +1,7 @@
from typing import Literal, Union, cast
from pydantic import BaseModel
import tweepy
from pydantic import BaseModel
from tweepy.client import Response
from backend.blocks.twitter._auth import (
@@ -26,34 +26,31 @@ from backend.blocks.twitter.tweepy_exceptions import handle_tweepy_exception
from backend.data.block import Block, BlockCategory, BlockOutput, BlockSchema
from backend.data.model import SchemaField
class UserId(BaseModel):
discriminator: Literal['user_id']
user_id: str = SchemaField(
description="The ID of the user to lookup",
default=""
)
discriminator: Literal["user_id"]
user_id: str = SchemaField(description="The ID of the user to lookup", default="")
class Username(BaseModel):
discriminator: Literal['username']
discriminator: Literal["username"]
username: str = SchemaField(
description="The Twitter username (handle) of the user",
default=""
description="The Twitter username (handle) of the user", default=""
)
class TwitterGetUserBlock(Block):
"""
Gets information about a single Twitter user specified by ID or username
"""
class Input(UserExpansionInputs):
credentials: TwitterCredentialsInput = TwitterCredentialsField(
["users.read", "offline.access"]
)
identifier: Union[UserId, Username] = SchemaField(
discriminator='discriminator',
discriminator="discriminator",
description="Choose whether to identify the user by their unique Twitter ID or by their username",
advanced=False,
)
@@ -79,10 +76,7 @@ class TwitterGetUserBlock(Block):
input_schema=TwitterGetUserBlock.Input,
output_schema=TwitterGetUserBlock.Output,
test_input={
"identifier": {
"discriminator": "username",
"username": "twitter"
},
"identifier": {"discriminator": "username", "username": "twitter"},
"credentials": TEST_CREDENTIALS_INPUT,
"expansions": None,
"tweet_fields": None,
@@ -136,7 +130,9 @@ class TwitterGetUserBlock(Block):
params = {
"id": identifier.user_id if isinstance(identifier, UserId) else None,
"username": identifier.username if isinstance(identifier, Username) else None,
"username": (
identifier.username if isinstance(identifier, Username) else None
),
"user_auth": False,
}
@@ -183,7 +179,7 @@ class TwitterGetUserBlock(Block):
input_data.identifier,
input_data.expansions,
input_data.tweet_fields,
input_data.user_fields
input_data.user_fields,
)
if id:
yield "id", id
@@ -200,7 +196,7 @@ class TwitterGetUserBlock(Block):
class UserIdList(BaseModel):
discriminator: Literal['user_id_list']
discriminator: Literal["user_id_list"]
user_ids: list[str] = SchemaField(
description="List of user IDs to lookup (max 100)",
placeholder="Enter user IDs",
@@ -208,8 +204,9 @@ class UserIdList(BaseModel):
advanced=False,
)
class UsernameList(BaseModel):
discriminator: Literal['username_list']
discriminator: Literal["username_list"]
usernames: list[str] = SchemaField(
description="List of Twitter usernames/handles to lookup (max 100)",
placeholder="Enter usernames",
@@ -217,6 +214,7 @@ class UsernameList(BaseModel):
advanced=False,
)
class TwitterGetUsersBlock(Block):
"""
Gets information about multiple Twitter users specified by IDs or usernames
@@ -228,7 +226,7 @@ class TwitterGetUsersBlock(Block):
)
identifier: Union[UserIdList, UsernameList] = SchemaField(
discriminator='discriminator',
discriminator="discriminator",
description="Choose whether to identify users by their unique Twitter IDs or by their usernames",
advanced=False,
)
@@ -256,7 +254,7 @@ class TwitterGetUsersBlock(Block):
test_input={
"identifier": {
"discriminator": "username_list",
"usernames": ["twitter", "twitterdev"]
"usernames": ["twitter", "twitterdev"],
},
"credentials": TEST_CREDENTIALS_INPUT,
"expansions": None,
@@ -312,8 +310,16 @@ class TwitterGetUsersBlock(Block):
)
params = {
"ids": ",".join(identifier.user_ids) if isinstance(identifier, UserIdList) else None,
"usernames": ",".join(identifier.usernames) if isinstance(identifier, UsernameList) else None,
"ids": (
",".join(identifier.user_ids)
if isinstance(identifier, UserIdList)
else None
),
"usernames": (
",".join(identifier.usernames)
if isinstance(identifier, UsernameList)
else None
),
"user_auth": False,
}

View File

@@ -1,9 +1,9 @@
/* flow.css or index.css */
body {
font-family: -apple-system, BlinkMacSystemFont, "Segoe UI", "Roboto", "Oxygen",
"Ubuntu", "Cantarell", "Fira Sans", "Droid Sans", "Helvetica Neue",
sans-serif;
font-family: -apple-system, BlinkMacSystemFont, "Segoe UI", "Roboto",
"Oxygen", "Ubuntu", "Cantarell", "Fira Sans", "Droid Sans",
"Helvetica Neue", sans-serif;
}
code {