Spaces:
Sleeping
Sleeping
File size: 2,764 Bytes
3cb34a2 911c0ac 3cb34a2 0508b3e 3cb34a2 0508b3e 3cb34a2 0508b3e 3cb34a2 0508b3e 3cb34a2 0508b3e 3cb34a2 ecbbafe 3cb34a2 0508b3e 3cb34a2 b151c5c |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 |
"""Manage parks table in the database."""
from sqlalchemy import text
from data_models.sql_connection import get_db_connection
class ParkManager:
def __init__(self):
"""Initialise connection and session."""
self.engine, self.session = get_db_connection()
def add_park(self, name) -> int:
query = text(
"""
INSERT INTO parks (name)
VALUES (:name)
RETURNING id
"""
)
try:
response = self.session.execute(query, {"name": name})
self.session.commit()
return response.all()[0][0]
except Exception as e:
self.session.rollback()
raise Exception(f"An error occurred while adding the park: {e}")
def get_parks(self):
"""
get all parks from the `parks` table.
Returns:
list[dict]: list of parks.
"""
try:
query = text("SELECT * FROM parks")
result = self.session.execute(query)
return [row._asdict() for row in result.fetchall()]
except Exception as e:
raise Exception(f"An error occurred while fetching the parks: {e}")
def get_park_id(self, park_name):
"""Get the park ID from the park name.
Args:
park_name (str): Name of the park.
"""
query = text("SELECT id FROM parks WHERE name = :park_name")
try:
result = self.session.execute(query, {"park_name": park_name}).fetchone()
return result[0] if result else None
except Exception as e:
raise Exception(f"An error occurred while getting the park ID: {e}")
def delete_park(self, park_id):
"""
Delete a park from the `parks` table.
Args:
park_id (int): ID of the park to delete.
Returns:
bool: True if the park was deleted, False otherwise.
"""
query = text("DELETE FROM parks WHERE id = :park_id")
try:
result = self.session.execute(query, {"park_id": park_id})
self.session.commit()
return result.rowcount > 0
except Exception as e:
self.session.rollback()
raise Exception(f"An error occurred while adding the park: {e}")
def close_connection(self):
"""Close the connection."""
self.session.close()
def get_park_count(self):
"""Get the number of parks in the database."""
query = text("SELECT COUNT(*) FROM parks")
try:
result = self.session.execute(query).fetchone()
return result[0]
except Exception as e:
raise Exception(f"An error occurred while getting the park count: {e}")
|