datapipelines 1.0.7

例子

例如，如果您的数据管道由内存缓存、数据库和 REST API 服务（按此顺序）组成，当您执行查询时，DataPipeline将首先查找内存缓存，然后查找数据库，然后在 REST API 中。如果在缓存中找到数据，则将其返回，并且不会查询数据库和 REST API。同样，如果在数据库中找到数据，则不会查询 REST API。

在数据源中找到数据后，数据会沿数据管道从其来源传回。沿途遇到的任何数据接收器都将存储该数据。因此，继续上面的示例，如果您的查询是由 REST API 提供的，则返回的数据将存储在您的数据库中，然后存储在缓存中。数据接收器只会存储它支持的数据。

每个数据接收器都可以为其支持的每种数据类型定义过期期限，但这取决于要实现的特定数据接收器。

用法

下面的代码是简化的代码，用于说明将 SQL 数据库请求的 Word 文档转换为 PDF 文档的示例。

请注意，没有 PDF 文档存储在数据库中，但如果请求，数据管道仍然可以返回一个。

# The four classes below implement a simple DataPipeline. The code would need to be filled in by the user.

class WordDoc:
    ...

class PDF:
    ...

class SQLDatabase(DataSource, DataSink):
    @get.register(WordDoc)  # Tells the DataPipeline that this SQL database can provide a WordDoc
    def get_word_doc(query: Dict[str, Any]) -> WordDoc:
        """Returns a WordDoc from an SQL database based on the `filename` in the query."""

    @put.register(WordDoc)  # Tell the DataPipeline that this SQL database can store a WordDoc
    def put_word_doc(doc: WordDoc, query: Dict[str, Any]):
        """Stores the document in the SQL database using the query as an identifier."""

class DocumentTransformer(Transformer):
    @transform.register(WordDoc, PDF)  # Tells the DataPipeline that we know how to convert a WordDoc to a PDF
    def Word_to_PDF(doc: WordDoc) -> PDF:
        """Converts a WordDoc to a PDF and returns the PDF."""


# The line of code below can now be used to request a PDF.
# The WordDoc with the filename `find_me` will be pulled from the SQL database then converted to a PDF and returned to the user.
my_pdf = pipeline.get(PDF, query={"filename": "find_me"})

# Note also that because we implemented a `put(WordDoc)` method in the SQLDatabase that it will also store WordDocs that pass through the SQL database via the pipeline but are not already in the database.