airbyte_cdk.sources.declarative.spec 

1#
2# Copyright (c) 2023 Airbyte, Inc., all rights reserved.
3#
4
5from airbyte_cdk.sources.declarative.spec.spec import Spec
6
7__all__ = ["Spec"]
@dataclass
class Spec: 
17@dataclass
18class Spec:
19    """
20    Returns a connection specification made up of information about the connector and how it can be configured
21
22    Attributes:
23        connection_specification (Mapping[str, Any]): information related to how a connector can be configured
24        documentation_url (Optional[str]): The link the Airbyte documentation about this connector
25    """
26
27    connection_specification: Mapping[str, Any]
28    parameters: InitVar[Mapping[str, Any]]
29    documentation_url: Optional[str] = None
30    advanced_auth: Optional[AuthFlow] = None
31
32    def generate_spec(self) -> ConnectorSpecification:
33        """
34        Returns the connector specification according the spec block defined in the low code connector manifest.
35        """
36
37        obj: dict[str, Mapping[str, Any] | str | AdvancedAuth] = {
38            "connectionSpecification": self.connection_specification
39        }
40
41        if self.documentation_url:
42            obj["documentationUrl"] = self.documentation_url
43        if self.advanced_auth:
44            self.advanced_auth.auth_flow_type = self.advanced_auth.auth_flow_type.value  # type: ignore # We know this is always assigned to an AuthFlow which has the auth_flow_type field
45            # Map CDK AuthFlow model to protocol AdvancedAuth model
46            obj["advanced_auth"] = self.advanced_auth.dict()
47
48        # We remap these keys to camel case because that's the existing format expected by the rest of the platform
49        return ConnectorSpecificationSerializer.load(obj)

Returns a connection specification made up of information about the connector and how it can be configured

Attributes:
  • connection_specification (Mapping[str, Any]): information related to how a connector can be configured
  • documentation_url (Optional[str]): The link the Airbyte documentation about this connector
Spec( connection_specification: Mapping[str, Any], parameters: dataclasses.InitVar[typing.Mapping[str, typing.Any]], documentation_url: Optional[str] = None, advanced_auth: Optional[airbyte_cdk.sources.declarative.models.declarative_component_schema.AuthFlow] = None)
connection_specification: Mapping[str, Any]
parameters: dataclasses.InitVar[typing.Mapping[str, typing.Any]]
documentation_url: Optional[str] = None
def generate_spec( self) -> airbyte_protocol_dataclasses.models.airbyte_protocol.ConnectorSpecification: 
32    def generate_spec(self) -> ConnectorSpecification:
33        """
34        Returns the connector specification according the spec block defined in the low code connector manifest.
35        """
36
37        obj: dict[str, Mapping[str, Any] | str | AdvancedAuth] = {
38            "connectionSpecification": self.connection_specification
39        }
40
41        if self.documentation_url:
42            obj["documentationUrl"] = self.documentation_url
43        if self.advanced_auth:
44            self.advanced_auth.auth_flow_type = self.advanced_auth.auth_flow_type.value  # type: ignore # We know this is always assigned to an AuthFlow which has the auth_flow_type field
45            # Map CDK AuthFlow model to protocol AdvancedAuth model
46            obj["advanced_auth"] = self.advanced_auth.dict()
47
48        # We remap these keys to camel case because that's the existing format expected by the rest of the platform
49        return ConnectorSpecificationSerializer.load(obj)

Returns the connector specification according the spec block defined in the low code connector manifest.