146 lines
4.1 KiB
Python
146 lines
4.1 KiB
Python
"""
|
|
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()
|