使用Python调用Gitlab User List API并存储到Amazon S3的优化需求

核心改进点 #

1. 处理GitLab API分页逻辑，拉取全量用户
2. 优化S3存储的JSON格式，提升可读性
3. 实现Lambda自动化调度与CloudFormation编排

一、全量用户拉取代码实现 #

GitLab API默认分页返回数据（每页最多100条），通过循环请求+分页判断即可获取所有用户：

import requests
import json
import boto3
import os

# 从Lambda环境变量读取配置，避免硬编码敏感信息
GITLAB_API_URL = os.environ["GITLAB_API_URL"]
GITLAB_ACCESS_TOKEN = os.environ["GITLAB_ACCESS_TOKEN"]
S3_TARGET_BUCKET = os.environ["S3_TARGET_BUCKET"]
S3_FILE_KEY = "gitlab_users/formatted_users.json"

def fetch_all_users():
    all_users = []
    current_page = 1
    per_page = 100  # 最大化每页数据量，减少请求次数

    while True:
        headers = {"Authorization": f"Bearer {GITLAB_ACCESS_TOKEN}"}
        params = {"page": current_page, "per_page": per_page}
        
        # 请求GitLab用户列表API
        response = requests.get(f"{GITLAB_API_URL}/users", headers=headers, params=params)
        response.raise_for_status()  # 抛出HTTP请求错误，便于CloudWatch日志排查
        
        page_users = response.json()
        if not page_users:
            break  # 无数据则终止循环
        
        all_users.extend(page_users)
        
        # 检查是否存在下一页（通过Link响应头判断）
        link_header = response.headers.get("Link")
        if not link_header or 'rel="next"' not in link_header:
            break
        
        current_page += 1

    return all_users

def lambda_handler(event, context):
    try:
        users = fetch_all_users()
        # 格式化JSON：缩进2空格、排序键、兼容非序列化类型
        formatted_json = json.dumps(users, indent=2, sort_keys=True, default=str)
        
        # 上传到S3
        s3_client = boto3.client("s3")
        s3_client.put_object(
            Bucket=S3_TARGET_BUCKET,
            Key=S3_FILE_KEY,
            Body=formatted_json,
            ContentType="application/json"
        )

        return {
            "statusCode": 200,
            "body": json.dumps(f"同步完成：共拉取{len(users)}个GitLab用户")
        }
    except Exception as e:
        return {
            "statusCode": 500,
            "body": json.dumps(f"同步失败：{str(e)}")
        }

二、CloudFormation编排与自动化调度 #

以下CloudFormation模板可一键创建Lambda函数、IAM权限、每周触发的EventBridge规则：

AWSTemplateFormatVersion: '2010-09-09'
Description: 每周同步私有GitLab用户列表到S3的自动化栈

Parameters:
  GitLabApiUrl:
    Type: String
    Description: 私有GitLab实例的API根地址（如https://gitlab.example.com/api/v4）
  GitLabAccessToken:
    Type: String
    Description: GitLab访问令牌（建议后续改用Secrets Manager存储）
  TargetS3Bucket:
    Type: String
    Description: 存储用户列表的S3桶名称

Resources:
  # Lambda执行角色
  LambdaExecutionRole:
    Type: AWS::IAM::Role
    Properties:
      AssumeRolePolicyDocument:
        Version: '2012-10-17'
        Statement:
          - Effect: Allow
            Principal: { Service: lambda.amazonaws.com }
            Action: sts:AssumeRole
      ManagedPolicyArns:
        - arn:aws:iam::aws:policy/service-role/AWSLambdaBasicExecutionRole
      Policies:
        - PolicyName: S3WriteAccess
          PolicyDocument:
            Version: '2012-10-17'
            Statement:
              - Effect: Allow
                Action: s3:PutObject
                Resource: !Sub "arn:aws:s3:::${TargetS3Bucket}/*"

  # GitLab用户同步Lambda函数
  GitLabUserSyncLambda:
    Type: AWS::Lambda::Function
    Properties:
      Runtime: python3.11
      Handler: index.lambda_handler
      Code:
        ZipFile: |
          import requests
          import json
          import boto3
          import os

          GITLAB_API_URL = os.environ["GITLAB_API_URL"]
          GITLAB_ACCESS_TOKEN = os.environ["GITLAB_ACCESS_TOKEN"]
          S3_TARGET_BUCKET = os.environ["S3_TARGET_BUCKET"]
          S3_FILE_KEY = "gitlab_users/formatted_users.json"

          def fetch_all_users():
              all_users = []
              current_page = 1
              per_page = 100

              while True:
                  headers = {"Authorization": f"Bearer {GITLAB_ACCESS_TOKEN}"}
                  params = {"page": current_page, "per_page": per_page}
                  response = requests.get(f"{GITLAB_API_URL}/users", headers=headers, params=params)
                  response.raise_for_status()
                  
                  page_users = response.json()
                  if not page_users:
                      break
                  
                  all_users.extend(page_users)
                  
                  link_header = response.headers.get("Link")
                  if not link_header or 'rel="next"' not in link_header:
                      break
                  
                  current_page += 1

              return all_users

          def lambda_handler(event, context):
              try:
                  users = fetch_all_users()
                  formatted_json = json.dumps(users, indent=2, sort_keys=True, default=str)
                  
                  s3_client = boto3.client("s3")
                  s3_client.put_object(
                      Bucket=S3_TARGET_BUCKET,
                      Key=S3_FILE_KEY,
                      Body=formatted_json,
                      ContentType="application/json"
                  )

                  return {
                      "statusCode": 200,
                      "body": json.dumps(f"同步完成：共拉取{len(users)}个GitLab用户")
                  }
              except Exception as e:
                  return {
                      "statusCode": 500,
                      "body": json.dumps(f"同步失败：{str(e)}")
                  }
      Environment:
        Variables:
          GITLAB_API_URL: !Ref GitLabApiUrl
          GITLAB_ACCESS_TOKEN: !Ref GitLabAccessToken
          S3_TARGET_BUCKET: !Ref TargetS3Bucket
      Role: !GetAtt LambdaExecutionRole.Arn

  # 每周同步的EventBridge规则（默认每周日凌晨0点执行）
  WeeklySyncRule:
    Type: AWS::Events::Rule
    Properties:
      ScheduleExpression: "cron(0 0 ? * SUN *)"
      Targets:
        - Arn: !GetAtt GitLabUserSyncLambda.Arn
          Id: GitLabUserSyncTarget

  # 允许EventBridge触发Lambda的权限
  LambdaInvokePermission:
    Type: AWS::Lambda::Permission
    Properties:
      FunctionName: !Ref GitLabUserSyncLambda
      Action: lambda:InvokeFunction
      Principal: events.amazonaws.com
      SourceArn: !GetAtt WeeklySyncRule.Arn

关键优化说明 #

• 分页逻辑：通过循环递增page参数，结合响应头Link判断是否存在下一页，确保无遗漏拉取所有用户
• JSON格式化：使用indent=2生成缩进结构，sort_keys=True统一字段顺序，default=str处理GitLab返回的日期等特殊类型
• 安全规范：敏感信息通过环境变量传递，后续可迁移至AWS Secrets Manager实现更安全的密钥管理
• 自动化调度：通过EventBridge的cron表达式配置每周执行，可根据需求调整触发时间（如cron(0 12 ? * FRI *)表示每周五中午12点）
• 错误排查：Lambda集成CloudWatch日志，response.raise_for_status()会抛出HTTP错误，便于快速定位API请求问题

内容的提问来源于stack exchange，提问作者ZookOG