Spaces:
Sleeping
Sleeping
| """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}") | |