Skip to content

[fix] 아티스트 수집 지연 대응 — is_coming 중복호출 제거·배치 트랜잭션 정리 #83

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
You-Hyuk merged 7 commits into from
Sep 9, 2026
Merged

[fix] 아티스트 수집 지연 대응 — is_coming 중복호출 제거·배치 트랜잭션 정리 #83

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
7 commits
Select commit Hold shift + click to select a range
ca1679b
[fix] update_artist_is_coming 중복 호출 제거
You-Hyuk Sep 6, 2026
a938ab6
[refactor] 배치 저장 함수 항목별 독립 트랜잭션으로 전환
You-Hyuk Sep 6, 2026
8da50fe
[chore] 배치 잡 실행 소요시간 로깅 추가
You-Hyuk Sep 6, 2026
ba90268
[test] _job_timer 소요시간 로깅 회귀 테스트 추가
You-Hyuk Sep 6, 2026
6d1361e
[fix] DB 레벨 statement_timeout/lock_timeout 설정 추가
You-Hyuk Sep 6, 2026
7c7e644
[fix] _job_timer가 monotonic 시계를 사용하도록 변경
You-Hyuk Sep 7, 2026
bc4edb0
[test] 배치 저장 함수 4종 트랜잭션 격리 회귀 테스트 추가
You-Hyuk Sep 7, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 5 additions & 1 deletion db/connection.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,7 +17,11 @@
f"@{os.getenv('DB_HOST')}:{os.getenv('DB_PORT')}/{os.getenv('DB_NAME')}"
)

engine = create_engine(_DATABASE_URL, pool_pre_ping=True)
engine = create_engine(
_DATABASE_URL,
pool_pre_ping=True,
connect_args={"options": "-c statement_timeout=10000 -c lock_timeout=5000"},
)
SessionLocal = sessionmaker(bind=engine, autocommit=False, autoflush=False)


Expand Down
214 changes: 127 additions & 87 deletions db/repository.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -302,44 +302,54 @@ def get_completed_concerts() -> list[dict]:


def save_setlists(setlists: list[dict]) -> None:
"""수집된 셋리스트를 setlist·setlist_track 테이블에 저장한다. 중복 시 무시."""
with get_session() as session:
for item in setlists:
row = session.execute(
text("""
INSERT INTO setlist (concert_id, setlist_fm_id, attribution_url, collected_at)
VALUES (:concert_id, :setlist_fm_id, :attribution_url, NOW())
ON CONFLICT (setlist_fm_id) DO NOTHING
RETURNING id
"""),
{
"concert_id": item["concert_id"],
"setlist_fm_id": item["setlist_fm_id"],
"attribution_url": item.get("attribution_url"),
},
).fetchone()

if not row or not item.get("tracks"):
continue

setlist_id = row[0]
session.execute(
text("""
INSERT INTO setlist_track (setlist_id, position, song_name, info)
VALUES (:setlist_id, :position, :song_name, :info)
"""),
[
"""수집된 셋리스트를 setlist·setlist_track 테이블에 저장한다. 중복 시 무시.

셋리스트별 독립 트랜잭션을 사용해 한 건 실패가 전체에 영향을 주지 않는다.
"""
saved = 0
for item in setlists:
try:
with get_session() as session:
row = session.execute(
text("""
INSERT INTO setlist
(concert_id, setlist_fm_id, attribution_url, collected_at)
VALUES (:concert_id, :setlist_fm_id, :attribution_url, NOW())
ON CONFLICT (setlist_fm_id) DO NOTHING
RETURNING id
"""),
{
"setlist_id": setlist_id,
"position": track["position"],
"song_name": track["song_name"],
"info": track.get("info"),
}
for track in item["tracks"]
],
"concert_id": item["concert_id"],
"setlist_fm_id": item["setlist_fm_id"],
"attribution_url": item.get("attribution_url"),
},
).fetchone()

if row and item.get("tracks"):
setlist_id = row[0]
session.execute(
text("""
INSERT INTO setlist_track (setlist_id, position, song_name, info)
VALUES (:setlist_id, :position, :song_name, :info)
"""),
[
{
"setlist_id": setlist_id,
"position": track["position"],
"song_name": track["song_name"],
"info": track.get("info"),
}
for track in item["tracks"]
],
)
saved += 1
except SQLAlchemyError as e:
logger.error(
"셋리스트 저장 실패 — 건너뜀: setlist_fm_id=%s, 오류=%s",
item.get("setlist_fm_id"), e,
)

logger.info("셋리스트 저장 완료: %d건 처리", len(setlists))
logger.info("셋리스트 저장 완료: %d / %d건 처리", saved, len(setlists))


def get_all_aliases() -> list[dict]:
Expand Down Expand Up @@ -619,40 +629,62 @@ def get_unmatched_concerts() -> list[dict]:


def save_concert_artists(matches: list[dict]) -> None:
"""매칭 결과를 concert_artist 테이블에 저장한다."""
with get_session() as session:
for match in matches:
session.execute(
text("""
INSERT INTO concert_artist (concert_id, artist_id)
VALUES (:concert_id, :artist_id)
ON CONFLICT (concert_id, artist_id) DO NOTHING
"""),
{
"concert_id": match["concert_id"],
"artist_id": match["artist_id"],
},
"""매칭 결과를 concert_artist 테이블에 저장한다.

매칭별 독립 트랜잭션을 사용해 한 건 실패가 전체에 영향을 주지 않는다.
"""
saved = 0
for match in matches:
try:
with get_session() as session:
session.execute(
text("""
INSERT INTO concert_artist (concert_id, artist_id)
VALUES (:concert_id, :artist_id)
ON CONFLICT (concert_id, artist_id) DO NOTHING
"""),
{
"concert_id": match["concert_id"],
"artist_id": match["artist_id"],
},
)
saved += 1
except SQLAlchemyError as e:
logger.error(
"공연-아티스트 매칭 저장 실패 — 건너뜀: concert_id=%s, artist_id=%s, 오류=%s",
match["concert_id"], match["artist_id"], e,
)
logger.info("공연-아티스트 매칭 저장 완료: %d건 처리", len(matches))
logger.info("공연-아티스트 매칭 저장 완료: %d / %d건 처리", saved, len(matches))


def save_concert_artist_candidates(matches: list[dict]) -> None:
"""자동 매칭 결과를 concert_artist_candidate 테이블에 저장한다. 중복 시 무시."""
with get_session() as session:
for match in matches:
session.execute(
text("""
INSERT INTO concert_artist_candidate
(concert_id, artist_id)
VALUES (:concert_id, :artist_id)
ON CONFLICT (concert_id, artist_id) DO NOTHING
"""),
{
"concert_id": match["concert_id"],
"artist_id": match["artist_id"],
},
"""자동 매칭 결과를 concert_artist_candidate 테이블에 저장한다. 중복 시 무시.

매칭별 독립 트랜잭션을 사용해 한 건 실패가 전체에 영향을 주지 않는다.
"""
saved = 0
for match in matches:
try:
with get_session() as session:
session.execute(
text("""
INSERT INTO concert_artist_candidate
(concert_id, artist_id)
VALUES (:concert_id, :artist_id)
ON CONFLICT (concert_id, artist_id) DO NOTHING
"""),
{
"concert_id": match["concert_id"],
"artist_id": match["artist_id"],
},
)
saved += 1
except SQLAlchemyError as e:
logger.error(
"공연-아티스트 후보 저장 실패 — 건너뜀: concert_id=%s, artist_id=%s, 오류=%s",
match["concert_id"], match["artist_id"], e,
)
logger.info("공연-아티스트 후보 저장 완료: %d건 처리", len(matches))
logger.info("공연-아티스트 후보 저장 완료: %d / %d건 처리", saved, len(matches))



Expand Down Expand Up @@ -722,31 +754,39 @@ def update_concert_fetch_attempted(concert_id: int) -> None:


def update_concert_status(concerts: list[dict]) -> None:
"""updatedate 변화 감지 시 status와 kopis_update_date를 갱신한다."""
"""updatedate 변화 감지 시 status와 kopis_update_date를 갱신한다.

공연별 독립 트랜잭션을 사용해 한 건 실패가 전체에 영향을 주지 않는다.
"""
updated = 0
with get_session() as session:
for concert in concerts:
row = session.execute(
text("SELECT kopis_update_date FROM concert WHERE kopis_id = :kopis_id"),
{"kopis_id": concert["kopis_id"]},
).fetchone()
for concert in concerts:
try:
with get_session() as session:
row = session.execute(
text("SELECT kopis_update_date FROM concert WHERE kopis_id = :kopis_id"),
{"kopis_id": concert["kopis_id"]},
).fetchone()

if row is None:
continue
if row is None:
continue

if row[0] != concert["updatedate"]:
session.execute(
text("""
UPDATE concert
SET status = :status, kopis_update_date = :kopis_update_date
WHERE kopis_id = :kopis_id
"""),
{
"status": _KOPIS_STATUS_MAP.get(concert["prfstate"], "PENDING"),
"kopis_update_date": concert["updatedate"],
"kopis_id": concert["kopis_id"],
},
)
updated += 1
if row[0] != concert["updatedate"]:
session.execute(
text("""
UPDATE concert
SET status = :status, kopis_update_date = :kopis_update_date
WHERE kopis_id = :kopis_id
"""),
{
"status": _KOPIS_STATUS_MAP.get(concert["prfstate"], "PENDING"),
"kopis_update_date": concert["updatedate"],
"kopis_id": concert["kopis_id"],
},
)
updated += 1
except SQLAlchemyError as e:
logger.error(
"공연 상태 갱신 실패 — 건너뜀: kopis_id=%s, 오류=%s", concert["kopis_id"], e
)

logger.info("공연 상태 갱신 완료: %d건 변경", updated)
Loading
Loading