26Z07e

argparse 라이브러리를 적용하여 Command Line 매개변수(Bucket, Prefix, DB 접속 정보)를 전달받아 실행할 수 있도록 작성한 aa.py 전체 코드입니다.

import argparse
import json
import sys
import pymysql


def get_prefix_metadata(
    conn, bucket: str, prefix: str
) -> dict:
    """StarRocks View(v_prefix_summary)를 조회하여 Prefix 및 하위 재귀 집계 결과를 반환합니다."""
    prefix_clean = prefix.strip('/')

    # Prefix가 빈 값(루트)인 경우 전체 버킷 집계, 값이 있는 경우 하위 재귀 포함 조건 적용
    if not prefix_clean:
        sql = """
            SELECT 
                bucket,
                '' AS target_prefix,
                SUM(total_object_count) AS total_object_count,
                SUM(total_size_bytes) AS total_size_bytes
            FROM v_prefix_summary
            WHERE bucket = %s
            GROUP BY bucket;
        """
        params = (bucket,)
    else:
        sql = """
            SELECT 
                bucket,
                %s AS target_prefix,
                SUM(total_object_count) AS total_object_count,
                SUM(total_size_bytes) AS total_size_bytes
            FROM v_prefix_summary
            WHERE bucket = %s
              AND (prefix = %s OR prefix LIKE CONCAT(%s, '/%%'))
            GROUP BY bucket;
        """
        params = (prefix_clean, bucket, prefix_clean, prefix_clean)

    with conn.cursor(pymysql.cursors.DictCursor) as cursor:
        cursor.execute(sql, params)
        result = cursor.fetchone()

    if not result:
        return {
            'bucket': bucket,
            'target_prefix': prefix_clean,
            'total_object_count': 0,
            'total_size_bytes': 0,
        }

    # SUM 결과 타입 정리 (None 처리 및 int 캐스팅)
    result['total_object_count'] = int(result['total_object_count'] or 0)
    result['total_size_bytes'] = int(result['total_size_bytes'] or 0)
    return result


def main():
    parser = argparse.ArgumentParser(
        description='StarRocks MinIO Prefix Metadata Aggregator'
    )

    # 필수 매개변수
    parser.add_argument(
        '-b',
        '--bucket',
        required=True,
        type=str,
        help='Target MinIO Bucket Name (Required)',
    )

    # 선택 매개변수
    parser.add_argument(
        '-p',
        '--prefix',
        default='',
        type=str,
        help="Target Prefix Path (Optional, default: '')",
    )
    parser.add_argument(
        '--host',
        default='localhost',
        type=str,
        help='StarRocks FE Host IP (default: localhost)',
    )
    parser.add_argument(
        '-P',
        '--port',
        default=9030,
        type=int,
        help='StarRocks MySQL Port (default: 9030)',
    )
    parser.add_argument(
        '-u',
        '--user',
        default='root',
        type=str,
        help='StarRocks User (default: root)',
    )
    parser.add_argument(
        '--password',
        default='',
        type=str,
        help="StarRocks Password (default: '')",
    )
    parser.add_argument(
        '--db',
        default='minio_metadata',
        type=str,
        help='Database Name (default: minio_metadata)',
    )

    args = parser.parse_args()

    # DB 연결 및 실행
    try:
        conn = pymysql.connect(
            host=args.host,
            port=args.port,
            user=args.user,
            password=args.password,
            db=args.db,
            charset='utf8mb4',
            autocommit=True,
        )

        metadata = get_prefix_metadata(
            conn=conn, bucket=args.bucket, prefix=args.prefix
        )

        # 결과를 JSON 형태로 출력
        print(json.dumps(metadata, indent=2, ensure_ascii=False))

    except Exception as e:
        print(f'[Error] Query Execution Failed: {e}', file=sys.stderr)
        sys.exit(1)
    finally:
        if 'conn' in locals() and conn.open:
            conn.close()


if __name__ == '__main__':
    main()

---

실행 예시

• 기본 실행 (특정 Bucket과 Prefix 지정)

python aa.py -b my-bucket -p aoo

• 하위 경로 조회 및 DB IP/포트 직접 입력

python aa.py --bucket my-bucket --prefix aoo/t1 --host 192.168.1.100 --port 9030 --user root

• 도움말(Help) 출력

python aa.py --help