tap_feed/streams.py
def schema(self) -> dict:
        """Create a dynamic schema based on the tap configuration provided."""
        feed_properties: list = [
            th.Property("feed_url", th.StringType),
        ]
        feed_properties.extend(
            [
                th.Property(f"feed_{name}", th.StringType)
                for name in self.config["feed_fields"]
            ]
        )
        feed_properties.extend(
            [
                th.Property("entry_id", th.StringType),
                th.Property(FeedStream.replication_key, th.DateTimeType),
            ]
        )
        feed_properties.extend(
            [
                th.Property(f"entry_{name}", th.StringType)
                for name in self.config["feed_entry_fields"]
                if name not in ["id", "published", "updated"]
            ]
        )
        return th.PropertiesList(*feed_properties).to_dict()
Create a dynamic schema based on the tap configuration provided.
Create a dynamic schema based on the tap configuration provided.
Create a dynamic schema based on the tap configuration provided .
def schema(self) -> dict:
        feed_properties: list = [
            th.Property("feed_url", th.StringType),
        ]
        feed_properties.extend(
            [
                th.Property(f"feed_{name}", th.StringType)
                for name in self.config["feed_fields"]
            ]
        )
        feed_properties.extend(
            [
                th.Property("entry_id", th.StringType),
                th.Property(FeedStream.replication_key, th.DateTimeType),
            ]
        )
        feed_properties.extend(
            [
                th.Property(f"entry_{name}", th.StringType)
                for name in self.config["feed_entry_fields"]
                if name not in ["id", "published", "updated"]
            ]
        )
        return th.PropertiesList(*feed_properties).to_dict()
def schema ( self ) -> dict : feed_properties : list = [ th . Property ( "feed_url" , th . StringType ) , ] feed_properties . extend ( [ th . Property ( f"feed_{name}" , th . StringType ) for name in self . config [ "feed_fields" ] ] ) feed_properties . extend ( [ th . Property ( "entry_id" , th . StringType ) , th . Property ( FeedStream . replication_key , th . DateTimeType ) , ] ) feed_properties . extend ( [ th . Property ( f"entry_{name}" , th . StringType ) for name in self . config [ "feed_entry_fields" ] if name not in [ "id" , "published" , "updated" ] ] ) return th . PropertiesList ( * feed_properties ) . to_dict ( )
Create a dynamic schema based on the tap configuration provided.
Create a dynamic schema based on the tap configuration provided .
"""Create a dynamic schema based on the tap configuration provided."""
self