小编mpa*_*tek的帖子

Argo 允许根据先前步骤的输出动态生成并行工作流程步骤。

这里提供了动态工作流程生成的示例： https: //github.com/argoproj/argo-workflows/blob/master/examples/loops-param-result.yaml

我正在尝试创建一个类似的工作流程，其中包含最终的“扇入”步骤，该步骤将从动态创建的并行步骤中读取输出。这是一个尝试：

apiVersion: argoproj.io/v1alpha1
kind: Workflow
metadata:
  generateName: loops-param-result-
spec:
  entrypoint: loop-param-result-example
  templates:
  - name: loop-param-result-example
    steps:
    - - name: generate
        template: gen-number-list
    - - name: write
        template: output-number
        arguments:
          parameters:
          - name: number
            value: "{{item}}"
        withParam: "{{steps.generate.outputs.result}}"
    - - name: fan-in
        template: fan-in
        arguments:
          parameters:
          - name: numbers
            value: "{{steps.write.outputs.parameters.number}}"

  - name: gen-number-list
    script:
      image: python:alpine3.6
      command: [python]
      source: |
        import json
        import sys
        json.dump([i for i in range(20, 31)], sys.stdout)

  - name: output-number
    inputs: …

在Python 3.6.1中,我尝试在io.TextIOWrapper中包装tempfile.SpooledTemporaryFile:

with tempfile.SpooledTemporaryFile() as tfh:
    do_some_download(tfh)
    tfh.seek(0)
    wrapper = io.TextIOWrapper(tfh, encoding='utf-8')
    yield from do_some_text_formatting(wrapper)

该行wrapper = io.TextIOWrapper(tfh, encoding='utf-8')给了我一个错误:

AttributeError: 'SpooledTemporaryFile' object has no attribute 'readable'

如果我创建一个简单的类这样的,我可以绕过错误(我得到类似的错误writable和seekable):

class MySpooledTempfile(tempfile.SpooledTemporaryFile):                                                                                
    @property                                                                                                                          
    def readable(self):                                                                                                                
        return self._file.readable                                                                                                     

    @property                                                                                                                          
    def writable(self):                                                                                                                
        return self._file.writable                                                                                                     

    @property                                                                                                                          
    def seekable(self):                                                                                                                
        return self._file.seekable 

tempfile.SpooledTemporaryFile有没有一个很好的理由还没有这些属性？