Add New Connector

1
class GhibliAuthConfig(AuthConfig):
2
    """Studio Ghibli authentication credentials schema."""
3
4
    # No authentication needed for this public API
5
    pass

1
class GhibliConfig(SourceConfig):
2
    """Studio Ghibli configuration schema."""
3
4
    include_rt_scores: bool = Field(
5
        default=True,
6
        title="Include RT Scores",
7
        description="Whether to include Rotten Tomatoes scores in the metadata"
8
    )

1
from datetime import datetime
2
from typing import List, Optional
3
4
from pydantic import Field
5
6
from airweave.platform.entities._base import ChunkEntity
7
8
class GhibliFilmEntity(ChunkEntity):
9
    """Schema for a Studio Ghibli film."""
10
11
    film_id: str = Field(..., description="Unique ID of the film")
12
    title: str = Field(..., description="Title of the film")
13
    original_title: str = Field(..., description="Original Japanese title")
14
    director: str = Field(..., description="Director of the film")
15
    producer: str = Field(..., description="Producer of the film")
16
    release_date: str = Field(..., description="Release date")
17
    running_time: str = Field(..., description="Running time in minutes")
18
    rt_score: Optional[str] = Field(None, description="Rotten Tomatoes score")
19
    people: List[str] = Field(default_factory=list, description="Characters in the film")
20
    species: List[str] = Field(default_factory=list, description="Species in the film")
21
    locations: List[str] = Field(default_factory=list, description="Locations in the film")
22
    vehicles: List[str] = Field(default_factory=list, description="Vehicles in the film")

1
from typing import AsyncGenerator, Optional, Dict, Any
2
import httpx
3
4
from airweave.platform.entities.ghibli import GhibliFilmEntity
5
from airweave.platform.decorators import source
6
from airweave.platform.sources._base import BaseSource
7
from airweave.platform.auth.schemas import AuthType
8
9
@source(
10
    name="Studio Ghibli",
11
    short_name="ghibli",
12
    auth_type=AuthType.none,
13
    auth_config_class="GhibliAuthConfig",
14
    config_class="GhibliConfig",
15
    labels=["Entertainment", "API"]
16
)
17
class GhibliSource(BaseSource):
18
    """Studio Ghibli source implementation."""
19
20
    @classmethod
21
    async def create(
22
        cls,
23
        credentials=None,
24
        config: Optional[Dict[str, Any]] = None
25
    ) -> "GhibliSource":
26
        """Create a new Ghibli source instance."""
27
        instance = cls()
28
        instance.config = config or {}
29
        return instance
30
31
    async def generate_entities(self) -> AsyncGenerator[GhibliFilmEntity, None]:
32
        """Generate entities from the Ghibli API."""
33
        async with httpx.AsyncClient() as client:
34
            response = await client.get("https://ghibli.rest/films")
35
            response.raise_for_status()
36
            films = response.json()
37
38
            for film in films:
39
                yield GhibliFilmEntity(
40
                    entity_id=film["id"],
41
                    film_id=film["id"],
42
                    title=film["title"],
43
                    original_title=film["original_title"],
44
                    content=film["description"],  # Required by ChunkEntity
45
                    director=film["director"],
46
                    producer=film["producer"],
47
                    release_date=film["release_date"],
48
                    running_time=film["running_time"],
49
                    rt_score=film["rt_score"] if self.config.get("include_rt_scores", True) else None,
50
                    people=film.get("people", []),
51
                    species=film.get("species", []),
52
                    locations=film.get("locations", []),
53
                    vehicles=film.get("vehicles", [])
54
                )