Skip to content
Snippets Groups Projects
Commit ff3be6e3 authored by Kenechukwu Orjiene's avatar Kenechukwu Orjiene
Browse files

Update 9 files

- /models/__init__.py
- /models/activities.py
- /models/actors.py
- /models/collections.py
- /models/fields.py
- /models/links.py
- /models/objects.py
- /serializers/__init__.py
- /serializers/json_serializer.py
parent 609da7f8
Branches
No related tags found
No related merge requests found
from .actors import Actor
from .objects import Object
from .activities import Activity
__all__ = ["Actor", "Object", "Activity"]
from pydantic import Field, root_validator
from typing import List, Union, Optional
from models.objects import APObject
from .fields import (
ActorField,
InstrumentField,
ObjectField,
OriginField,
ResultField,
TargetField,
TypeField,
AnyOfField,
ClosedField,
OneOfField
)
class APActivity(APObject):
actor: Union[ActorField, List[ActorField]]
object: Union[ObjectField, List[ObjectField]]
target: Union[TargetField, List[TargetField]]
origin: Union[OriginField, List[OriginField]]
result: Union[ResultField, List[ResultField]]
instrument: Union[InstrumentField, List[InstrumentField]]
class APIntransitiveActivity(APActivity):
object = None
class APIgnore(APActivity):
type = TypeField = Field("Ignore", const=True)
class APOffer(ActorField):
type = TypeField = Field("Offer", const=True)
class APAccept(APActivity):
type: TypeField = Field("Accept", const=True)
class APAdd(APActivity):
type: TypeField = Field("Add", const=True)
class APAnnounce(APActivity):
type: TypeField = Field("Announce", const=True)
class APArrivve(APIntransitiveActivity):
type: TypeField = Field("Arrive", const=True)
class APBlock(APIgnore):
type: TypeField = Field("Block", const=True)
class APCreate(APActivity):
type: TypeField = Field("Create", const=True)
class APDelete(APActivity):
type: TypeField = Field("Delete", const=True)
class APDislike(APActivity):
type: TypeField = Field("Dislike", const=True)
class APFlag(APActivity):
type: TypeField = Field("Flag", const=True)
class APFollow(APActivity):
type: TypeField = Field("Follow", const=True)
class APInvite(APOffer):
type: TypeField = Field("Invite", const=True)
class APJoin(APActivity):
type: TypeField = Field("Join", const=True)
class APLeave(APActivity):
type: TypeField = Field("Leave", const=True)
class APLike(APActivity):
type: TypeField = Field("Like", const=True)
class APListen(APActivity):
type: TypeField = Field("Listen", const=True)
class APMove(APActivity):
type: TypeField = Field("Move", const=True)
class APRead(APActivity):
type: TypeField = Field("Read", const=True)
class APReject(APActivity):
type: TypeField = Field("Reject", const=True)
class APRemove(APActivity):
type: TypeField = Field("Remove", const=True)
class APTentativeAccept(APActivity):
type: TypeField = Field("TentativeAccept", const=True)
class APTentativeReject(APActivity):
type: TypeField = Field("TentativeReject", const=True)
class APTravel(APActivity):
type: TypeField = Field("Travel", const=True)
class APUndo(APActivity):
type: TypeField = Field("Undo", const=True)
class APUpdate(APActivity):
type: TypeField = Field("Update", const=True)
class APView(APActivity):
type: TypeField = Field("View", const=True)
class Question(APIntransitiveActivity):
type: TypeField = Field("Question", const=True)
oneOf: Optional[List[OneOfField]] = None
anyOf: Optional[List[AnyOfField]] = None
closed: Optional[ClosedField] = None
@root_validator(pre=True)
def check_any_of_or_one_of(cls, values):
"""Ensure that either 'anyOf' or 'oneOf' is used, but not both."""
if values.get('anyOf') and values.get('oneOf'):
raise ValueError("A Question object must not have both 'anyOf' and 'oneOf'.")
return values
class APQuestionAnyOf(Question):
oneOf: None
class APQuestionOneOf(Question):
anyOf: None
class APQuestionNone(Question):
oneOf: None
anyOf: None
from __future__ import annotations
from pydantic import Field
from typing import Optional
from .objects import APObject
from .fields import (
FollowersField,
FollowingField,
InboxField,
LikedField,
OutboxField,
PreferredUsernameMapField,
PreferredUsernameField,
StreamsField,
EndpointsField,
TypeField
)
class APActor(APObject):
"""
Base class for ActivityPub actors.
"""
inbox: InboxField
outbox: OutboxField
following: Optional[FollowingField] = None
followers: Optional[FollowersField] = None
liked: Optional[LikedField] = None
streams: Optional[StreamsField] = None
preferred_username: Optional[PreferredUsernameField] = None
preferred_username_map: Optional[PreferredUsernameMapField] = None
endpoints: Optional[EndpointsField] = None
class APPerson(APActor):
"""
Represents a Person actor in ActivityPub.
"""
type: TypeField = Field("Person", const=True)
class APGroup(APActor):
"""
Represents a Group actor in ActivityPub.
"""
type: TypeField = Field("Group", const=True)
class APOrganization(APActor):
"""
Represents an Organization actor in ActivityPub.
"""
type: TypeField = Field("Organization", const=True)
class APApplication(APActor):
"""
Represents an Application actor in ActivityPub.
"""
type: TypeField = Field("Application", const=True)
class APService(APActor):
"""
Represents a Service actor in ActivityPub.
"""
type: TypeField = Field("Service", const=True)
\ No newline at end of file
from __future__ import annotations
from typing import Optional, List, Union
from pydantic import Field
from .fields import (
CollectionCurrentField,
CollectionFirstField, CollectionItemsField,
CollectionLastField,
TotalItemsField,
CollectionPageNextField,
CollectionPagePartOfField,
CollectionPagePrevField,
StartIndexField,
OrderedCollectionItemsField,
TypeField
)
from .objects import APObject
class APCollection(APObject):
"""
Collection object, inheriting from Object
"""
type: TypeField = Field("Collection", const=True)
items: List[CollectionItemsField]
total_items: Optional[TotalItemsField] = None
first: Optional[CollectionFirstField] = None
last: Optional[CollectionLastField] = None
current: Optional[CollectionCurrentField] = None
class APCollectionPage(APCollection):
type: TypeField = Field("CollectionPage", const=True)
part_of: Optional[CollectionPagePartOfField] = None
next: Optional[CollectionPageNextField] = None
prev: Optional[CollectionPagePrevField] = None
class APOrderedCollection(APCollectionPage):
type: TypeField = Field("OrderedCollectionPage", const=True)
ordered_items: OrderedCollectionItemsField
class APOrderedCollectionPage(APCollection, APCollectionPage):
type: TypeField = Field("OrderedCollectionPage", const=True)
start_index: StartIndexField
\ No newline at end of file
from pydantic import BaseModel, HttpUrl, Field, validator
from typing import Union, Dict, Optional, Literal, List
from datetime import datetime
from .links import APLink, APMention
from .collections import (
APCollection,
APOrderedCollection,
APOrderedCollectionPage,
APCollectionPage
)
from .objects import (
APObject,
APEvent,
APPlace,
APRelationship,
APProfile,
APTombstone,
APArticle,
APAudio,
APDocument,
APImage,
APNote,
APPage,
APVideo
)
class AnyCollection(BaseModel):
__root__: Union[APCollection, APOrderedCollection]
# AnyAPObject is a union of many APObject types, including AnyCollection
class AnyAPObject(BaseModel):
__root__: Union[
APObject,
APArticle,
APAudio,
APDocument,
APEvent,
APImage,
APNote,
APPage,
APPlace,
APProfile,
APRelationship,
APTombstone,
APVideo,
AnyCollection
]
class DateTime(BaseModel):
__root__: Union[str | datetime]
class LanguageTag(BaseModel):
__root__: str
class MediaType(BaseModel):
__root__: str
@validator("__root__")
def validate_media_type(cls, value):
"""
Validation helper for MIME types.
This validates the type of media file in use
it recognizes "image/jpeg", "image/png", "application/json", "text/html"
"""
allowed_mimes = ["image/jpeg", "image/png", "application/json", "text/html"]
if value and value not in allowed_mimes:
raise ValueError(f"Invalid MIME type: {value}")
return value
class Duration(BaseModel):
__root__: str
class ContextField(BaseModel):
__root__: Union[str, Dict[str, str]] = Field(alias="@context")
class UrlField(BaseModel):
__root__: Union[str, APLink]
class IdField(BaseModel):
__root__: str
class TypeField(BaseModel):
__root__: str
class DurationField(BaseModel):
__root__: Duration
class ContentField(BaseModel):
__root__: str
class ContentMapField(BaseModel):
__root__: Dict[str, ContentField]
class MediaTypeField(BaseModel):
__root__: MediaType
class NameField(BaseModel):
__root__: str
class NameMapField(BaseModel):
__root__: Dict[str, NameField]
class SummaryField(BaseModel):
__root__: str
class SummaryMapField(BaseModel):
__root__: Dict[str, SummaryField]
class EndTimeField(BaseModel):
__root__: DateTime
class StartTimeField(BaseModel):
__root__: DateTime
class PublishedField(BaseModel):
__root__: DateTime
class UpdatedField(BaseModel):
__root__: DateTime
class RepliesField(BaseModel):
__root__: Union[str, AnyCollection]
class RelationshipField(BaseModel):
__root__: Union[str, AnyAPObject]
class IconField(BaseModel):
__root__: Union[str, APImage, APLink]
class ImageField(BaseModel):
__root__: Union[str, APImage, APLink]
class AttachmentField(BaseModel):
__root__: Union[str, AnyAPObject, APLink]
class AudienceField(BaseModel):
__root__: Union[str, AnyAPObject, APLink]
class InReplyToField(BaseModel):
__root__: Union[str, AnyAPObject, APLink]
class LocationField(BaseModel):
__root__: Union[str, APPlace, APLink]
class PreviewField(BaseModel):
__root__: Union[str, AnyAPObject, APLink]
class ToField(BaseModel):
__root__: Union[str, AnyAPObject, APLink]
class BtoField(BaseModel):
__root__: Union[str, AnyAPObject, APLink]
class CcField(BaseModel):
__root__: Union[str, AnyAPObject, APLink]
class BccField(BaseModel):
__root__: Union[str, AnyAPObject, APLink]
class GeneratorField(BaseModel):
__root__: Union[str, AnyAPObject, APLink]
class ObjectField(BaseModel):
__root__: Union[str, AnyAPObject, APLink]
class AttributedToField(BaseModel):
__root__: Union[str, AnyAPObject, APLink, APMention]
class TagField(BaseModel):
__root__: Union[str, AnyAPObject, APLink, APMention]
class ActorField(BaseModel):
__root__: Union[str, AnyAPObject, APLink]
class TargetField(BaseModel):
__root__: Union[str, AnyAPObject, APLink]
class ResultField(BaseModel):
__root__: Union[str, AnyAPObject, APLink]
class OriginField(BaseModel):
__root__: Union[str, AnyAPObject, APLink]
class InstrumentField(BaseModel):
__root__: Union[str, AnyAPObject, APLink]
class OneOfField(BaseModel):
__root__: Union[str, AnyAPObject, APLink]
class AnyOfField(BaseModel):
__root__: Union[str, AnyAPObject, APLink]
class SubjectField(BaseModel):
__root__: Union[str, AnyAPObject, APLink]
class AccuracyField(BaseModel):
__root__: float
class AltitudeField(BaseModel):
__root__: float
class LatitudeField(BaseModel):
__root__: float
class LongitudeField(BaseModel):
__root__: float
class RadiusField(BaseModel):
__root__: float = Field(..., ge=0.0)
class UnitsField(BaseModel):
__root__: Union[Literal['cm', 'feet', 'inches', 'km', 'm', 'miles'], str]
class ClosedField(BaseModel):
__root__: Union[str, AnyAPObject, APLink, DateTime, bool]
class FormerTypeField(BaseModel):
__root__: str
class DeletedField(BaseModel):
__root__: DateTime
class DescribesField(BaseModel):
__root__: Union[str, AnyAPObject]
class HrefField(BaseModel):
__root__: HttpUrl
class HreflangField(BaseModel):
__root__: LanguageTag
class RelField(BaseModel):
__root__: List[str]
class HeightField(BaseModel):
__root__: int = Field(..., ge=0)
class WidthField(BaseModel):
__root__: int = Field(..., ge=0)
class TotalItemsField(BaseModel):
__root__: int = Field(..., ge=0)
class CollectionCurrentField(BaseModel):
__root__: Union[str, APCollectionPage, APLink]
class CollectionFirstField(BaseModel):
__root__: Union[str, APCollectionPage, APLink]
class CollectionLastField(BaseModel):
__root__: Union[str, APCollectionPage, APLink]
class CollectionItemsField(BaseModel):
__root__: Union[str, APCollectionPage, APLink]
class CollectionPagePartOfField(BaseModel):
__root__: Union[str, APCollection, APLink]
class CollectionPageNextField(BaseModel):
__root__: Union[str, APCollection, APLink]
class CollectionPagePrevField(BaseModel):
__root__: Union[str, APCollection, APLink]
class OrderedCollectionCurrentField(BaseModel):
__root__: Union[str, APOrderedCollectionPage, APLink]
class OrderedCollectionFirstField(BaseModel):
__root__: Union[str, APOrderedCollectionPage, APLink]
class OrderedCollectionLastField(BaseModel):
__root__: Union[str, APOrderedCollectionPage, APLink]
class OrderedCollectionItemsField(BaseModel):
__root__: Union[str, APOrderedCollectionPage, APLink]
class OrderedCollectionPagePartOfField(BaseModel):
__root__: Union[str, APOrderedCollection, APLink]
class OrderedCollectionPageNextField(BaseModel):
__root__: Union[str, APOrderedCollection, APLink]
class OrderedCollectionPagePrevField(BaseModel):
__root__: Union[str, APOrderedCollection, APLink]
class StartIndexField(BaseModel):
__root__: int = Field(..., ge=0)
class SourceField(BaseModel):
content: str
media_type: MediaType
class InboxField(BaseModel):
__root__: Union[str, APOrderedCollection, APLink]
class OutboxField(BaseModel):
__root__: Union[str, APOrderedCollection, APLink]
class FollowingField(BaseModel):
__root__: Union[str, APCollection, APOrderedCollection, APLink]
class FollowersField(BaseModel):
__root__: Union[str, APCollection, APOrderedCollection, APLink]
class LikedField(BaseModel):
__root__: Union[str, APCollection, APOrderedCollection, APLink]
class LikesField(BaseModel):
__root__: Union[str, APCollection, APOrderedCollection, APLink]
class SharesField(BaseModel):
__root__: Union[str, APCollection, APOrderedCollection, APLink]
class StreamsField(BaseModel):
__root__: Union[str, APCollection, APOrderedCollection, APLink]
class PreferredUsernameField(BaseModel):
__root__: str
class PreferredUsernameMapField(BaseModel):
__root__: Dict[str, str]
class ProxyUrlField(BaseModel):
__root__: str
class OauthAuthorizationEndpointField(BaseModel):
__root__: str
class OauthTokenEndpointField(BaseModel):
__root__: str
class ProvideClientKeyField(BaseModel):
__root__: str
class SignClientKeyField(BaseModel):
__root__: str
class SharedInboxField(BaseModel):
__root__: str
class EndpointsField(BaseModel):
proxy_url: Optional[ProxyUrlField] = None
oauth_authorization_endpoint: Optional[OauthAuthorizationEndpointField] = None
oauth_token_endpoint: Optional[OauthTokenEndpointField] = None
provide_client_key: Optional[ProvideClientKeyField] = None
sign_client_key: Optional[SignClientKeyField] = None
shared_inbox: Optional[SharedInboxField] = None
from __future__ import annotations
from pydantic import Field
from typing import Optional, List, Union
from serializers.json_serializer import ActivityPubModel
from .fields import (
HeightField,
HreflangField,
HrefField,
MediaTypeField,
NameMapField,
NameField, PreviewField,
RelField, TypeField, WidthField,
ContextField,
)
class APLink(ActivityPubModel):
"""
Base class for links
"""
context: Optional[ContextField] = Field(alias="@context")
type: Union[TypeField, List[TypeField]]
name: Optional[NameField] = None
name_map: Optional[NameMapField] = None
href: HrefField
hreflang: HreflangField
rel: Optional[Union[RelField, List[RelField]]] = None
height: Optional[HeightField] = None
width: Optional[WidthField] = None
preview: Optional[PreviewField] = None
media_type: Optional[MediaTypeField]
class APMention(APLink):
"""
"""
type: TypeField = Field("Mention", const=True)
\ No newline at end of file
from __future__ import annotations
from pydantic import HttpUrl, Field, AnyUrl, PositiveInt, validator
from typing import Optional
from serializers.json_serializer import ActivityPubModel
from .fields import (
UrlField,
IconField,
ImageField,
AttachmentField,
AudienceField,
InReplyToField,
LocationField,
PreviewField,
ToField,
BtoField,
CcField,
BccField,
AttributedToField,
TagField,
GeneratorField,
IdField,
TypeField,
ContentField,
ContentMapField,
MediaTypeField,
NameField,
NameMapField,
EndTimeField,
StartTimeField,
PublishedField,
RepliesField,
SummaryField,
SummaryMapField,
UpdatedField,
DurationField,
ContextField,
AccuracyField,
AltitudeField,
LatitudeField,
LongitudeField,
RadiusField,
UnitsField,
DescribesField,
SubjectField,
ObjectField,
RelationshipField,
FormerTypeField,
DeletedField,
SourceField,
LikesField,
SharesField,
)
class _APObjectBaseModel(ActivityPubModel):
"""
Base class for all objects in the ActivityStreams vocabulary
"""
id: Optional[IdField] = None
type: TypeField
context: Optional[ContextField] = Field(alias="@context")
name: Optional[NameField] = None
# class Image(_MediaObject):
# pass
class APObject(_APObjectBaseModel):
"""
The Object class represents any object in the vocabulary.
https://www.w3.org/TR/activitystreams-core/#object
https://www.w3.org/TR/activitystreams-vocabulary/#object-types
"""
attachment: Optional[AttachmentField] = None
attributed_to: Optional[AttributedToField] = None
audience: Optional[AudienceField] = None
bcc: Optional[BccField] = None
bto: Optional[BtoField] = None
cc: Optional[CcField] = None
content: Optional[ContentField] = None
content_map: Optional[ContentMapField] = None
generator: Optional[GeneratorField] = None
icon: Optional[IconField] = None
image: Optional[ImageField] = None
in_reply_to: Optional[InReplyToField] = None
location: Optional[LocationField] = None
end_time: Optional[EndTimeField] = None
preview: Optional[PreviewField] = None
published: Optional[PublishedField] = None
replies: Optional[RepliesField] = None
start_time: Optional[StartTimeField] = None
summary: Optional[SummaryField] = None
summary_map: Optional[SummaryMapField] = None
name: Optional[NameField] = None
name_map: Optional[NameMapField] = None
tag: Optional[TagField] = None
updated: Optional[UpdatedField] = None
url: Optional[UrlField] = None
to: Optional[ToField] = None
duration: Optional[DurationField] = None
media_type: Optional[MediaTypeField] = None
# added these field based on the typescript definitions,
# TODO yet to verify if this should be included or not
source: Optional[SourceField] = None
likes: Optional[LikesField] = None
shares: Optional[SharesField] = None
@validator("end_time", "start_time")
def validate_time_constraints(cls, v, values, field):
if field.name == "end_time" and v and values.get("start_time") and v < values["start_time"]:
raise ValueError("end_time cannot be before start_time")
return v
class APEvent(APObject):
"""
Represents any kind of event.
"""
type: TypeField = Field("Event", const=True)
class APPlace(APObject):
"""
Represents a logical or physical location.
https://www.w3.org/TR/activitystreams-vocabulary/#dfn-place
https://www.w3.org/TR/activitystreams-vocabulary/#places
"""
type: TypeField = Field("Place", const=True)
accuracy: AccuracyField
altitude: AltitudeField
latitude: LatitudeField
logitude: LongitudeField
raduis: RadiusField
units: UnitsField
@validator("units")
def validate_unit(cls, value):
allowed_units = ["cm", " feet", " inches", " km", " m", " miles" ]
if value and value not in allowed_units:
raise ValueError(f"Invalid units type: {value}")
return value
class APProfile(APObject):
"""
Represents Profile
"""
type: TypeField = Field("Profile", const=True)
describes: DescribesField
class APRelationship(APObject):
type: TypeField = Field("Relationship", const=True)
subject: SubjectField
object: ObjectField
relationship: RelationshipField
class APTombstone(APObject):
type: TypeField = Field("Tombstone", const=True)
former_type: FormerTypeField
deleted: DeletedField
class APArticle(APObject):
type: TypeField = Field("Article", const=True)
class APAudio(APObject):
type: TypeField = Field("Audio", const=True)
class APDocument(APObject):
type: TypeField = Field("Document", const=True)
class APImage(APObject):
type: TypeField = Field("Image", const=True)
class APNote(APObject):
type: TypeField = Field("Note", const=True)
class APPage(APObject):
type: TypeField = Field("Page", const=True)
class APVideo(APObject):
type: TypeField = Field("Video", const=True)
\ No newline at end of file
from .json_serializer import ActivityPubModel
__all__ = ["ActivityPubModel"]
\ No newline at end of file
import warnings
from pydantic import BaseModel, root_validator
from typing import Dict, Union, Generator, Callable, Any
def to_camel_case(snake_str: str) -> str:
"""
Converts snake_case to camelCase
"""
pascal_string = "".join(x.capitalize() for x in snake_str.lower().split("_"))
return snake_str[0].lower() + pascal_string[1:]
class ActivityPubModel(BaseModel):
"""
A robust class combining serialization, deserialization, and object type handling
for ActivityPub models.
"""
_subtypes_: Dict[str, type] = dict()
class Config:
alias_generator = to_camel_case
@root_validator(pre=True)
@classmethod
def set_object_type(cls, values: Dict[str, str]) -> Dict[str, str]:
"""
Reads the type of an object from its class and adds the class to the value
so that it appears in the serialized representation.
"""
values["type"] = cls.__name__
return values
def __init_subclass__(cls) -> None:
if cls.__name__ in cls._subtypes_:
warnings.warn(
f"A subclass called {cls.__name__} is already initialized. Unexpected behavior might occur."
)
cls._subtypes_[cls.__name__] = cls
@classmethod
def __get_validators__(cls) -> Generator[Callable[..., None], None, None]:
yield cls._convert_to_real_type
@classmethod
def _convert_to_real_type(cls, data: Union[str, Dict[str, Any]]) -> Any:
"""
Ascertains the data type of a deserialized object. This ensures the correct object is created.
"""
if isinstance(data, str):
raise NotImplementedError("Actor needs dereferencing, not yet supported")
if isinstance(data, dict):
data_type = data.get("type")
if data_type is None:
raise ValueError("Missing 'type' in ActivityPub Model")
sub = cls._subtypes_.get(data_type)
if sub is None:
raise TypeError(f"Unsupported sub-type: {data_type}")
return sub(**data)
@classmethod
def parse_obj(cls, obj: Union[str, Dict[str, Any]]) -> Any:
"""
Deserializes a string into a Python object.
"""
return cls._convert_to_real_type(obj)
def dict(self, *args: Any, **kwargs: Any) -> Dict[str, Any]:
"""
Converts an object to a Python dict with by_alias=True and exclude_none=True by default.
"""
if "by_alias" not in kwargs:
kwargs.update(by_alias=True)
if "exclude_none" not in kwargs:
kwargs.update(exclude_none=True)
return super().dict(*args, **kwargs)
@classmethod
def serialize(cls, obj: BaseModel) -> str:
"""
Serializes an object into a JSON string.
"""
return obj.json(indent=4)
@classmethod
def deserialize(cls, json_str: str, model: BaseModel) -> BaseModel:
"""
Deserializes a JSON string into the appropriate model object.
"""
return model.parse_raw(json_str)
0% Loading or .
You are about to add 0 people to the discussion. Proceed with caution.
Please register or to comment