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

Revert "Merge branch 'main' into 'main'" 

This reverts merge request !2
parent 91db4190
No related branches found
No related tags found
1 merge request!3Definition of Objects, Actors, and Activity Model (reverted)
Show whitespace changes
Inline Side-by-side
Showing
with 58 additions and 1154 deletions
README.md
+ 58
− 234
View file @ ee15095b
# Building an ActivityPub Library for Python Developers # pyfed
## Introduction
ActivityPub is a protocol for social networking that is decentralized enabling the interchange of data between different social media systems. Thus, its aim of this library is to create the ActivityPub protocol in the form of Python language to the benefit of python programmers who wish to incorporate ActivityPub in their Apps. With that in mind, it is clear that ActivityPub would transform our online interactions for the better as it encourages users to leave the centralized, client-server model for more decentralized, federated networks. This library will appeal to developer of decentralized social networks, enthusiastic studying the federated protocols and all those who would like to develop social applications.
## Approach
To build a robust and reliable ActivityPub library, we will follow these steps:
1. **Model Definition**
- Define Python classes for ActivityPub entities (e.g., Actor, Object, Activity) using Pydantic for data validation and serialization.
**Example Model Definition**
```python
from pydantic import BaseModel, HttpUrl
from typing import List, Optional
class Actor(BaseModel):
id: HttpUrl
type: str
name: str
inbox: HttpUrl
outbox: HttpUrl
class Object(BaseModel):
id: HttpUrl
type: str
attributedTo: Actor
to: List[Actor] = []
content: str
class Activity(BaseModel):
id: HttpUrl
type: str
actor: Actor
object: Object
to: List[Actor] = []
context: Optional[str] = "https://www.w3.org/ns/activitystreams"
actor = Actor(
id="https://chatty.example/ben/",
type="Person",
name="Ben",
inbox="https://chatty.example/ben/inbox",
outbox="https://chatty.example/ben/outbox"
)
obj = Object(
id="https://example.com/message",
type="Note",
attributedTo=actor,
to=[actor],
content="Hello, World!"
)
...
```
2. **Serialization and Deserialization**
- Implement methods to convert Pydantic models to and from JSON-LD format, which is used by ActivityStreams.
- Pydantic provides built-in methods for serialization and deserialization, simplifying the process.
```python
from pydantic import BaseModel
from typing import Any
class JsonSerializer:
def serialize(self, obj: BaseModel) -> str:
return obj.json(indent=4)
def deserialize(self, json_str: str, model: BaseModel) -> BaseModel: ## Getting started
return model.parse_raw(json_str)
serializer = JsonSerializer() To make it easy for you to get started with GitLab, here's a list of recommended next steps.
# Serialize an activity Already a pro? Just edit this README.md and make it your own. Want to make it easy? [Use the template at the bottom](#editing-this-readme)!
serialized_activity = serializer.serialize(activity)
# Deserialize back to an Activity object ## Add your files
deserialized_activity = serializer.deserialize(serialized_activity, Activity)
```
3. **Server Endpoints** - [ ] [Create](https://docs.gitlab.com/ee/user/project/repository/web_editor.html#create-a-file) or [upload](https://docs.gitlab.com/ee/user/project/repository/web_editor.html#upload-a-file) files
- Design a client class for sending ActivityPub requests (e.g., `ActivityPubClient`). - [ ] [Add files using the command line](https://docs.gitlab.com/ee/gitlab-basics/add-file.html#add-a-file-using-the-command-line) or push an existing Git repository with the following command:
- Implement server endpoints for receiving ActivityPub requests (e.g., `ActivityPubServer`).
- Use a library like `aiohttp` for HTTP client functionality.
**Example Server Endpoint**
```python
from aiohttp import web
class ActivityPubServer:
async def handle_inbox_request(self, request: web.Request) -> web.Response:
# This method will parse the incoming JSON-LD,
# validate the sender's signature, and route the activity to the appropriate handler.
data = await request.json()
activity = Activity.parse_obj(data) # Validate and parse incoming data
# Process the incoming activity...
...
async def handle_outbox_request(self, request: web.Request) -> web.Response:
# This method will parse the outgoing activity and prepare it for delivery.
data = await request.json()
activity = Activity.parse_obj(data) # Validate and parse incoming data
# Prepare and send the outgoing activity...
...
app = web.Application()
app.add_routes([web.post("/inbox", ActivityPubServer().handle_inbox_request)])
app.add_routes([web.post("/outbox", ActivityPubServer().handle_outbox_request)])
web.run_app(app)
```
cd existing_repo
git remote add origin https://dev.funkwhale.audio/funkwhale/pyfed.git
git branch -M main
git push -uf origin main
``` ```
4. **Activity Handling** ## Integrate with your tools
- Implement activity handling logic, including:
- Activity creation and validation.
- Activity forwarding and delivery.
- Activity processing and updating.
**Example Activity Handling** - [ ] [Set up project integrations](https://dev.funkwhale.audio/funkwhale/pyfed/-/settings/integrations)
```python ## Collaborate with your team
class ActivityHandler:
def create_activity(self, actor: Actor, object: Object) -> Activity:
# Logic to create an activity
return Activity(id="some_id", type="Create", actor=actor, object=object, to=[actor])
def forward_activity(self, activity: Activity) -> None: - [ ] [Invite team members and collaborators](https://docs.gitlab.com/ee/user/project/members/)
# Logic to forward an activity - [ ] [Create a new merge request](https://docs.gitlab.com/ee/user/project/merge_requests/creating_merge_requests.html)
... - [ ] [Automatically close issues from merge requests](https://docs.gitlab.com/ee/user/project/issues/managing_issues.html#closing-issues-automatically)
- [ ] [Enable merge request approvals](https://docs.gitlab.com/ee/user/project/merge_requests/approvals/)
- [ ] [Set auto-merge](https://docs.gitlab.com/ee/user/project/merge_requests/merge_when_pipeline_succeeds.html)
def process_activity(self, activity: Activity) -> None: ## Test and Deploy
# Logic to process an activity
...
Use the built-in continuous integration in GitLab.
class CreateActivityHandler(ActivityHandler): - [ ] [Get started with GitLab CI/CD](https://docs.gitlab.com/ee/ci/quick_start/index.html)
def process_activity(self, activity: Activity) -> None: - [ ] [Analyze your code for known vulnerabilities with Static Application Security Testing (SAST)](https://docs.gitlab.com/ee/user/application_security/sast/)
# Logic specific to processing create activities - [ ] [Deploy to Kubernetes, Amazon EC2, or Amazon ECS using Auto Deploy](https://docs.gitlab.com/ee/topics/autodevops/requirements.html)
... - [ ] [Use pull-based deployments for improved Kubernetes management](https://docs.gitlab.com/ee/user/clusters/agent/)
- [ ] [Set up protected environments](https://docs.gitlab.com/ee/ci/environments/protected_environments.html)
class FollowActivityHandler(ActivityHandler): ***
def process_activity(self, activity: Activity) -> None:
# Logic specific to processing follow activities
...
class LikeActivityHandler(ActivityHandler): # Editing this README
def process_activity(self, activity: Activity) -> None:
# Logic specific to processing like activities
...
# Handlers would be selected dynamically based on activity type.
```
**Example Use Cases** When you're ready to make this README your own, just edit this file and use the handy template below (or feel free to structure it however you want - this is just a starting point!). Thanks to [makeareadme.com](https://www.makeareadme.com/) for this template.
- Sending a message from one actor to another
- Creating a new activity and forwarding it to other actors
- Processing incoming activities and updating the local state
**Library Structure** ## Suggestions for a good README
The library will be structured as follows:
``` Every project is different, so consider which of these sections apply to yours. The sections used in the template are suggestions for most open source projects. Also keep in mind that while a README can be too long and detailed, too long is better than too short. If you think your README is too long, consider utilizing another form of documentation rather than cutting out information.
pyfed/
│
├── __init__.py
│
├── models/ # Contains data models for ActivityPub entities
│ ├── __init__.py
│ ├── actor.py
│ ├── object.py
│ └── activity.py
│
├── serializers/ # Handles serialization and deserialization of models
│ ├── __init__.py
│ ├── json_serializer.py
│ ├── client.py
│ └── server.py
│
├── utils.py # Common utility functions and validation helpers
│
├── security/ # Implement security features like HTTP signatures
│ ├── __init__.py
│ └── http_signatures.py
│
├── tests/ # Test cases for models, serializers, client, and server
│ ├── __init__.py
│ ├── test_models.py
│ ├── test_serializers.py
│ ├── test_client.py
│ └── test_server.py
│
├── CONTRIBUTING.md
└── README.md
```
## Installation and Usage Instructions ## Name
To install the library, run Choose a self-explaining name for your project.
```bash ## Description
pip install pyfed Let people know what your project can do specifically. Provide context and add a link to any reference visitors might be unfamiliar with. A list of Features or a Background subsection can also be added here. If there are alternatives to your project, this is a good place to list differentiating factors.
```
**Basic Usage Example** ## Badges
Here’s a simple example of how to instantiate the server, create an actor and an object, and invoke an endpoint: On some READMEs, you may see small images that convey metadata, such as whether or not all the tests are passing for the project. You can use Shields to add some to your README. Many services also have instructions for adding a badge.
```python
from activitypub import ActivityPubServer, Actor, Object
# Initialize server
server = ActivityPubServer()
# Create an actor and an object
actor = Actor(
id="https://chatty.example/ben/",
type="Person",
name="Ben",
inbox="https://chatty.example/ben/inbox",
outbox="https://chatty.example/ben/outbox"
)
obj = Object(
id="https://example.com/message",
type="Note",
attributedTo=actor,
to=[actor],
content="Hello, World!"
)
# Create and serialize an activity
activity = server.create_activity(actor, obj)
serialized_activity = server.serialize(activity)
# (In a real scenario, you would send this serialized activity via a request)
```
## Contribution Guidelines ## Visuals
We welcome contributions to the library. Please see the `CONTRIBUTING.md` file for guidelines on how to contribute. You can report issues or request features on the [GitHub issues page](https://github.com/yourusername/pyfed/issues). Depending on what you are making, it can be a good idea to include screenshots or even a video (you'll frequently see GIFs rather than actual videos). Tools like ttygif can help, but check out Asciinema for a more sophisticated method.
## Dependencies ## Installation
The library depends on the following major dependencies: Within a particular ecosystem, there may be a common way of installing things, such as using Yarn, NuGet, or Homebrew. However, consider the possibility that whoever is reading your README is a novice and would like more guidance. Listing specific steps helps remove ambiguity and gets people to using your project as quickly as possible. If it only runs in a specific context like a particular programming language version or operating system or has dependencies that have to be installed manually, also add a Requirements subsection.
- `aiohttp` (version >=3.7.4) for HTTP client functionality.
- `pydantic` (version >=1.8.0) for data validation and serialization.
- `pytest` (version >=6.0) for testing.
## Future Plans ## Usage
We plan to implement additional features, including: Use examples liberally, and show the expected output if you can. It's helpful to have inline the smallest example of usage that you can demonstrate, while providing links to more sophisticated examples if they are too long to reasonably include in the README.
- Support for more ActivityPub endpoints (e.g., followers, following).
- Improved security and authentication mechanisms.
- Enhanced activity handling logic.
- Better error handling and debugging tools.
For a detailed roadmap of the project, see the `ROADMAP.md` file. ## Support
Tell people where they can go to for help. It can be any combination of an issue tracker, a chat room, an email address, etc.
## Additional Resources ## Roadmap
For more information on ActivityPub and its ecosystem, please see the following resources: If you have ideas for releases in the future, it is a good idea to list them in the README.
- [ActivityPub specification](https://www.w3.org/TR/activitypub/) ## Contributing
- [ActivityStreams standard](https://www.w3.org/TR/activitystreams-core/) State if you are open to contributions and what your requirements are for accepting them.
- [JSON-LD specification](https://www.w3.org/TR/json-ld/)
## License For people who want to make changes to your project, it's helpful to have some documentation on how to get started. Perhaps there is a script that they should run or some environment variables that they need to set. Make these steps explicit. These instructions could also be useful to your future self.
This project is licensed under the MIT License. See the `LICENSE` file for details.
## Timeline You can also document commands to lint the code or run tests. These steps help to ensure high code quality and reduce the likelihood that the changes inadvertently break something. Having instructions for running tests is especially helpful if it requires external setup, such as starting a Selenium server for testing in a browser.
**Review of Library by October**
**Completion by December**
--- ## Authors and acknowledgment
Show your appreciation to those who have contributed to the project.
This is just a starting point, and the approach may evolve as we progress with the implementation. We welcome contributions and feedback from the community to help shape the library into a robust and reliable implementation of the ActivityPub protocol. ## License
For open source projects, say how it is licensed.
![ActivityPub](flow.drawio.png) ## Project status
\ No newline at end of file If you have run out of energy or time for your project, put a note at the top of the README saying that development has slowed down or stopped completely. Someone may choose to fork your project or volunteer to step in as a maintainer or owner, allowing your project to keep going. You can also make an explicit request for maintainers.
__init__.py deleted 100644 → 0
+ 0
− 0
View file @ 91db4190
flow.drawio.png deleted 100644 → 0
+ 0
− 0
View file @ 91db4190
flow.drawio.png

22.1 KiB

models/__init__.py deleted 100644 → 0
+ 0
− 5
View file @ 91db4190
from .actors import Actor
from .objects import Object
from .activities import Activity
__all__ = ["Actor", "Object", "Activity"]
models/activities.py deleted 100644 → 0
+ 0
− 131
View file @ 91db4190
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
models/actors.py deleted 100644 → 0
+ 0
− 60
View file @ 91db4190
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
models/collections.py deleted 100644 → 0
+ 0
− 42
View file @ 91db4190
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
models/fields.py deleted 100644 → 0
+ 0
− 368
View file @ 91db4190
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
models/links.py deleted 100644 → 0
+ 0
− 36
View file @ 91db4190
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
models/objects.py deleted 100644 → 0
+ 0
− 183
View file @ 91db4190
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
serializers/__init__.py deleted 100644 → 0
+ 0
− 3
View file @ 91db4190
from .json_serializer import ActivityPubModel
__all__ = ["ActivityPubModel"]
\ No newline at end of file
serializers/json_serializer.py deleted 100644 → 0
+ 0
− 92
View file @ 91db4190
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% or .
You are about to add 0 people to the discussion. Proceed with caution.
Please register or to comment