""" Common Utility Functions 공통으로 사용되는 유틸리티 함수들을 정의합니다. 사용 예시: from app.utils.common import generate_task_id, generate_uuid # task_id 생성 task_id = await generate_task_id(session=session, table_name=Project) # uuid 생성 user_uuid = await generate_uuid(session=session, table_name=User) Note: 페이지네이션 기능은 app.utils.pagination 모듈을 사용하세요: from app.utils.pagination import PaginatedResponse, get_paginated """ import os import time from typing import Any, Optional, Type from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession def _generate_uuid7_string() -> str: """UUID7 문자열을 생성합니다. UUID7 구조 (RFC 9562): - 48 bits: Unix timestamp (밀리초) - 4 bits: 버전 (7) - 12 bits: 랜덤 - 2 bits: variant (10) - 62 bits: 랜덤 - 총 128 bits -> 36자 (하이픈 포함) Returns: 36자리 UUID7 문자열 (xxxxxxxx-xxxx-7xxx-yxxx-xxxxxxxxxxxx) """ # 현재 시간 (밀리초) timestamp_ms = int(time.time() * 1000) # 랜덤 바이트 (10바이트 = 80비트) random_bytes = os.urandom(10) # UUID7 바이트 구성 (16바이트 = 128비트) # 처음 6바이트: 타임스탬프 (48비트) uuid_bytes = timestamp_ms.to_bytes(6, byteorder="big") # 다음 2바이트: 버전(7) + 랜덤 12비트 # 0x7000 | (random 12 bits) rand_a = int.from_bytes(random_bytes[0:2], byteorder="big") version_rand = (0x7000 | (rand_a & 0x0FFF)).to_bytes(2, byteorder="big") uuid_bytes += version_rand # 다음 2바이트: variant(10) + 랜덤 62비트의 앞 6비트 # 0x80 | (random 6 bits) + random 8 bits rand_b = random_bytes[2] variant_rand = bytes([0x80 | (rand_b & 0x3F)]) + random_bytes[3:4] uuid_bytes += variant_rand # 나머지 6바이트: 랜덤 uuid_bytes += random_bytes[4:10] # 16진수로 변환 hex_str = uuid_bytes.hex() # UUID 형식으로 포맷팅 (8-4-4-4-12) return f"{hex_str[:8]}-{hex_str[8:12]}-{hex_str[12:16]}-{hex_str[16:20]}-{hex_str[20:32]}" async def generate_task_id( session: Optional[AsyncSession] = None, table_name: Optional[Type[Any]] = None, ) -> str: """고유한 task_id를 생성합니다. Args: session: SQLAlchemy AsyncSession (optional) table_name: task_id 컬럼이 있는 SQLAlchemy 테이블 클래스 (optional) Returns: str: 생성된 UUID7 문자열 (36자) Usage: # 단순 UUID7 생성 task_id = await generate_task_id() # 테이블에서 중복 검사 후 생성 task_id = await generate_task_id(session=session, table_name=Project) """ task_id = _generate_uuid7_string() if session is None or table_name is None: return task_id while True: result = await session.execute( select(table_name).where(table_name.task_id == task_id) ) existing = result.scalar_one_or_none() if existing is None: return task_id task_id = _generate_uuid7_string() async def generate_uuid( session: Optional[AsyncSession] = None, table_name: Optional[Type[Any]] = None, ) -> str: """고유한 UUID7을 생성합니다. Args: session: SQLAlchemy AsyncSession (optional) table_name: user_uuid 컬럼이 있는 SQLAlchemy 테이블 클래스 (optional) Returns: str: 생성된 UUID7 문자열 (36자) Usage: # 단순 UUID7 생성 new_uuid = await generate_uuid() # 테이블에서 중복 검사 후 생성 new_uuid = await generate_uuid(session=session, table_name=User) """ new_uuid = _generate_uuid7_string() if session is None or table_name is None: return new_uuid while True: result = await session.execute( select(table_name).where(table_name.user_uuid == new_uuid) ) existing = result.scalar_one_or_none() if existing is None: return new_uuid new_uuid = _generate_uuid7_string()