first commit
This commit is contained in:
291
repositories/stadium_repository.py
Normal file
291
repositories/stadium_repository.py
Normal file
@@ -0,0 +1,291 @@
|
||||
from db import get_connection
|
||||
|
||||
|
||||
def search_stadiums_for_admin(q: str = "") -> list[dict]:
|
||||
conn = get_connection()
|
||||
try:
|
||||
with conn.cursor() as cur:
|
||||
if q.strip():
|
||||
pattern = f"%{q.strip()}%"
|
||||
cur.execute(
|
||||
"""
|
||||
SELECT
|
||||
s.id,
|
||||
s.name,
|
||||
s.stadium_gfx,
|
||||
s.city,
|
||||
s.address,
|
||||
s.external_id
|
||||
FROM stadiums s
|
||||
WHERE
|
||||
s.name ILIKE %s
|
||||
OR COALESCE(s.stadium_gfx, '') ILIKE %s
|
||||
OR COALESCE(s.city, '') ILIKE %s
|
||||
OR COALESCE(s.address, '') ILIKE %s
|
||||
OR COALESCE(s.external_id, '') ILIKE %s
|
||||
ORDER BY s.name ASC, s.id ASC
|
||||
LIMIT 200
|
||||
""",
|
||||
(pattern, pattern, pattern, pattern, pattern),
|
||||
)
|
||||
else:
|
||||
cur.execute(
|
||||
"""
|
||||
SELECT
|
||||
s.id,
|
||||
s.name,
|
||||
s.stadium_gfx,
|
||||
s.city,
|
||||
s.address,
|
||||
s.external_id
|
||||
FROM stadiums s
|
||||
ORDER BY s.id DESC
|
||||
LIMIT 200
|
||||
"""
|
||||
)
|
||||
|
||||
rows = cur.fetchall()
|
||||
|
||||
return [
|
||||
{
|
||||
"id": row[0],
|
||||
"name": row[1] or "",
|
||||
"stadium_gfx": row[2] or "",
|
||||
"city": row[3] or "",
|
||||
"address": row[4] or "",
|
||||
"external_id": row[5] or "",
|
||||
}
|
||||
for row in rows
|
||||
]
|
||||
finally:
|
||||
conn.close()
|
||||
|
||||
|
||||
def get_stadium_by_id(stadium_id: int) -> dict | None:
|
||||
conn = get_connection()
|
||||
try:
|
||||
with conn.cursor() as cur:
|
||||
cur.execute(
|
||||
"""
|
||||
SELECT
|
||||
s.id,
|
||||
s.name,
|
||||
s.stadium_gfx,
|
||||
s.city,
|
||||
s.address,
|
||||
s.external_id
|
||||
FROM stadiums s
|
||||
WHERE s.id = %s
|
||||
LIMIT 1
|
||||
""",
|
||||
(stadium_id,),
|
||||
)
|
||||
row = cur.fetchone()
|
||||
|
||||
if not row:
|
||||
return None
|
||||
|
||||
return {
|
||||
"id": row[0],
|
||||
"name": row[1] or "",
|
||||
"stadium_gfx": row[2] or "",
|
||||
"city": row[3] or "",
|
||||
"address": row[4] or "",
|
||||
"external_id": row[5] or "",
|
||||
}
|
||||
finally:
|
||||
conn.close()
|
||||
|
||||
|
||||
def get_stadium_by_name(name: str):
|
||||
stadium_name = (name or "").strip()
|
||||
if not stadium_name:
|
||||
return None
|
||||
|
||||
conn = get_connection()
|
||||
try:
|
||||
with conn.cursor() as cur:
|
||||
cur.execute(
|
||||
"""
|
||||
SELECT
|
||||
s.id,
|
||||
s.name,
|
||||
s.stadium_gfx,
|
||||
s.city,
|
||||
s.address,
|
||||
s.external_id
|
||||
FROM stadiums s
|
||||
WHERE LOWER(s.name) = LOWER(%s)
|
||||
LIMIT 1
|
||||
""",
|
||||
(stadium_name,),
|
||||
)
|
||||
return cur.fetchone()
|
||||
finally:
|
||||
conn.close()
|
||||
|
||||
|
||||
def get_stadium_by_external_id(external_id: str):
|
||||
ext_id = (external_id or "").strip()
|
||||
if not ext_id:
|
||||
return None
|
||||
|
||||
conn = get_connection()
|
||||
try:
|
||||
with conn.cursor() as cur:
|
||||
cur.execute(
|
||||
"""
|
||||
SELECT
|
||||
s.id,
|
||||
s.name,
|
||||
s.stadium_gfx,
|
||||
s.city,
|
||||
s.address,
|
||||
s.external_id
|
||||
FROM stadiums s
|
||||
WHERE s.external_id = %s
|
||||
LIMIT 1
|
||||
""",
|
||||
(ext_id,),
|
||||
)
|
||||
return cur.fetchone()
|
||||
finally:
|
||||
conn.close()
|
||||
|
||||
|
||||
def create_stadium(
|
||||
name: str,
|
||||
stadium_gfx: str | None = None,
|
||||
external_id: str | None = None,
|
||||
city: str | None = None,
|
||||
address: str | None = None,
|
||||
) -> int:
|
||||
stadium_name = (name or "").strip()
|
||||
if not stadium_name:
|
||||
raise ValueError("Stadium name is required")
|
||||
|
||||
ext_value = (external_id or "").strip() or None
|
||||
city_value = (city or "").strip() or None
|
||||
address_value = (address or "").strip() or None
|
||||
gfx_value = (stadium_gfx or "").strip() or None
|
||||
|
||||
conn = get_connection()
|
||||
try:
|
||||
with conn.cursor() as cur:
|
||||
cur.execute(
|
||||
"""
|
||||
INSERT INTO stadiums (
|
||||
name,
|
||||
stadium_gfx,
|
||||
external_id,
|
||||
city,
|
||||
address,
|
||||
created_at,
|
||||
updated_at
|
||||
)
|
||||
VALUES (%s, %s, %s, %s, %s, NOW(), NOW())
|
||||
RETURNING id
|
||||
""",
|
||||
(
|
||||
stadium_name,
|
||||
gfx_value,
|
||||
ext_value,
|
||||
city_value,
|
||||
address_value,
|
||||
),
|
||||
)
|
||||
stadium_id = cur.fetchone()[0]
|
||||
|
||||
conn.commit()
|
||||
return stadium_id
|
||||
except Exception:
|
||||
conn.rollback()
|
||||
raise
|
||||
finally:
|
||||
conn.close()
|
||||
|
||||
|
||||
def get_or_create_stadium(
|
||||
name: str,
|
||||
external_id: str | None = None,
|
||||
city: str | None = None,
|
||||
address: str | None = None,
|
||||
) -> int | None:
|
||||
stadium_name = (name or "").strip()
|
||||
if not stadium_name:
|
||||
return None
|
||||
|
||||
ext_id = (external_id or "").strip()
|
||||
|
||||
if ext_id:
|
||||
existing_by_external = get_stadium_by_external_id(ext_id)
|
||||
if existing_by_external:
|
||||
return existing_by_external[0]
|
||||
|
||||
existing_by_name = get_stadium_by_name(stadium_name)
|
||||
if existing_by_name:
|
||||
return existing_by_name[0]
|
||||
|
||||
return create_stadium(
|
||||
name=stadium_name,
|
||||
stadium_gfx=stadium_name,
|
||||
external_id=ext_id or None,
|
||||
city=city,
|
||||
address=address,
|
||||
)
|
||||
|
||||
|
||||
def update_stadium_admin(
|
||||
stadium_id: int,
|
||||
stadium_gfx: str = "",
|
||||
city: str = "",
|
||||
address: str = "",
|
||||
external_id: str = "",
|
||||
) -> None:
|
||||
conn = get_connection()
|
||||
try:
|
||||
with conn.cursor() as cur:
|
||||
cur.execute(
|
||||
"""
|
||||
UPDATE stadiums
|
||||
SET
|
||||
stadium_gfx = %s,
|
||||
city = %s,
|
||||
address = %s,
|
||||
external_id = NULLIF(%s, ''),
|
||||
updated_at = NOW()
|
||||
WHERE id = %s
|
||||
""",
|
||||
(
|
||||
stadium_gfx.strip(),
|
||||
city.strip(),
|
||||
address.strip(),
|
||||
external_id.strip(),
|
||||
stadium_id,
|
||||
),
|
||||
)
|
||||
conn.commit()
|
||||
except Exception:
|
||||
conn.rollback()
|
||||
raise
|
||||
finally:
|
||||
conn.close()
|
||||
|
||||
|
||||
def get_stadium_display_name_by_id(stadium_id: int) -> str | None:
|
||||
conn = get_connection()
|
||||
try:
|
||||
with conn.cursor() as cur:
|
||||
cur.execute(
|
||||
"""
|
||||
SELECT COALESCE(NULLIF(stadium_gfx, ''), name)
|
||||
FROM stadiums
|
||||
WHERE id = %s
|
||||
LIMIT 1
|
||||
""",
|
||||
(stadium_id,),
|
||||
)
|
||||
row = cur.fetchone()
|
||||
return row[0] if row else None
|
||||
finally:
|
||||
conn.close()
|
||||
Reference in New Issue
Block a user