如何编写对 CSV 的 boto3 响应？答案

【问题标题】：How to write boto3 response to CSV?如何编写对 CSV 的 boto3 响应？
【发布时间】：2020-06-12 15:10:41
【问题描述】：

Python/Boto3 的新手，所以这有点令人困惑。我正在尝试使用 csv.writer 将 AWS Security Hub 调查结果写入 csv，但仅响应中的某些项目。我可以将正确的列和行写入 csv，但是当我尝试遍历 writer 时，它只会重复同一行，而不是响应中的其他数据。我觉得我忽略了一些简单的事情，感谢任何帮助。

def getSecurityHubFindings():
  hub = boto3.client('securityhub')
  findingsList = []
  for key in paginate(hub.get_findings, Filters=filters, PaginationConfig={'MaxItems': MAX_ITEMS}):
    scantype = key['Types']
    str1 = ''.join(scantype)
    port=key['ProductFields']['attributes:2/value']
    vgw=key['ProductFields']['attributes:3/value']
    scantype = key['Types']
    str1 = ''.join(scantype)
    findingAccountId = key['AwsAccountId']
    findingLastObservedAt=key['LastObservedAt']
    findingFirstObservedAt=key['FirstObservedAt']
    findingCreatedAt=key['CreatedAt']
    findingrecommendation=key['Remediation']['Recommendation']
    findingTypes=key['Types']
    InstanceId=key['Resources'][0]['Id']
    findingInstanceId=str(InstanceId)
    findingAppCode=key['Resources'][0]['Tags']['AppCode']
    findingGeneratorId=key['GeneratorId']
    findingProductArn=key['ProductArn']
    findingTitle=key['Title']
    findingsList.append(key)

    if (str1 == 'Software and Configuration Checks/AWS Security Best Practices/Network Reachability - Recognized port reachable from a Peered VPC'):
      vgw=''
      port=key['ProductFields'][ 'attributes:4/value']
      peeredvpc= key['ProductFields']['attributes:2/value']

    if (str1 == 'Software and Configuration Checks/AWS Security Best Practices/Network Reachability - Recognized port reachable from a Virtual Private Gateway'):
      peeredvpc=''
      sev = key['Severity']['Product']
      if (sev == 3):
        findingSeverity='LOW'
      elif (sev == 6):
        findingSeverity='MEDIUM'
      elif ( sev == 9):
        findingSeverity='HIGH'

    rows = [findingAccountId, findingGeneratorId, findingTitle,findingProductArn,findingSeverity,findingAppCode,findingFirstObservedAt,findingLastObservedAt,findingCreatedAt,findingrecommendation,findingTypes,port,vgw,peeredvpc,findingInstanceId]

    columns = ('Account ID', 'Generator ID', 'Title', 'Product ARN', 'Severity', 'AppCode', 'First Observed At','Last Observed At', 'Created At', 'Recommendation', 'Types', 'Port', 'VGW', 'Peered VPC', 'Instance #ID')

    with open(FILE_NAME, mode='w', newline='',) as writefile:
      writefile_writer = csv.writer(writefile, delimiter=',', quotechar='"', quoting=csv.QUOTE_ALL)
      writefile_writer.writerow(columns)
      i = 0
      while i < MAX_ITEMS:
        writefile_writer.writerow(rows)
        i +=1

  return(findingsList)

【问题讨论】：

您正在为每一行打开 CSV 文件，并且您正在使用 w 模式，因此每次都会删除该文件。打开文件，然后在您的上下文管理器中，遍历安全结果并一一写入。

标签： python amazon-web-services csv boto3

【解决方案1】：

一般流程应该是：

def getSecurityHubFindings():
    ...

    # Open output file and write header
    columns = ('Account ID', 'Generator ID', 'Title', 'Product ARN', 'Severity', 'AppCode', 'First Observed At','Last Observed At', 'Created At', 'Recommendation', 'Types', 'Port', 'VGW', 'Peered VPC', 'Instance #ID')

    with open(FILE_NAME, mode='w', newline='',) as writefile:
      writefile_writer = csv.writer(writefile, delimiter=',', quotechar='"', quoting=csv.QUOTE_ALL)
      writefile_writer.writerow(columns)

      ## Loop through response
      for key in paginate(...):

          ...
          (get data here)
          ...

          # Write output
          row = [findingAccountId, findingGeneratorId, findingTitle,findingProductArn,findingSeverity,findingAppCode,findingFirstObservedAt,findingLastObservedAt,findingCreatedAt,findingrecommendation,findingTypes,port,vgw,peeredvpc,findingInstanceId]
          writefile_writer.writerow(row)

【讨论】：

【解决方案2】：

您每次都使用“w”选项在 for 循环中打开文件，该选项会截断文件 [1] 并从头开始写入，因此每次都会覆盖 csv。

块

      while i < MAX_ITEMS:
        writefile_writer.writerow(rows)
        i +=1

似乎也错了，这只是将同一行（即使它称为行）写入 MAX_ITEMS 次。您可能希望打开 csv 文件并在 for 循环之外写入标题名称，然后为 for 循环的每次迭代写入一行。

【讨论】：