Spaces:
Sleeping
Sleeping
File size: 3,676 Bytes
911c0ac 3cb34a2 911c0ac 3cb34a2 911c0ac 3cb34a2 0508b3e 3cb34a2 0508b3e 3cb34a2 0508b3e 3cb34a2 0508b3e 3cb34a2 0508b3e 3cb34a2 0508b3e 3cb34a2 0508b3e 3cb34a2 ecbbafe 3cb34a2 ecbbafe 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 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 |
"""Manages images table in the database."""
from datetime import datetime
from sqlalchemy import text
from data_models.sql_connection import get_db_connection
class ImageManager:
"""ImageManager class."""
def __init__(self):
"""Initialise connection and session."""
self.engine, self.session = get_db_connection()
def add_image(self, name: str, created_at: datetime, park_id=None) -> int:
"""
Add an image to the `images` table.
Args:
name (str): Image name.
created_at (datetime): Image creation date.
park_id (int):id of the park where the image was taken.
Returns:
dict: Information of the added image.
"""
query = text(
"""
INSERT INTO images (name, created_at, park_id)
VALUES (:name, :created_at, :park_id)
RETURNING id
"""
)
try:
response = self.session.execute(
query, {"name": name, "created_at": created_at, "park_id": park_id}
).fetchone()
self.session.commit()
return response[0]
except Exception as e:
self.session.rollback()
raise Exception(f"An error occurred while adding the image: {e}")
def get_image_id(self, image_name):
"""
Get the image ID from the image name.
Args:
image_name (str): Name of the image.
Returns:
int: ID of the image.
"""
query = text("SELECT id FROM images WHERE name = :image_name")
try:
result = self.session.execute(query, {"image_name": image_name})
return result[0] if result else None
except Exception as e:
raise Exception(f"An error occurred while getting the image ID: {e}")
def get_images_by_park(self, park_id):
"""
Get all images from the `images` table.
Args:
park_id (int): ID of the park to filter images.
Returns:
list[dict]: List of images.
"""
query = text("SELECT * FROM images")
if park_id:
query = text("SELECT * FROM images WHERE park_id = :park_id")
try:
result = self.session.execute(
query, {"park_id": park_id} if park_id else {}
)
return [row._asdict() for row in result]
except Exception as e:
raise Exception(f"Erreur lors de la récupération des images : {e}")
def delete_image(self, image_id):
"""
Delete an image by its ID.
Args:
image_id (int): Image ID.
Returns:
bool: True if the deletion was successful, False otherwise.
"""
query = text("DELETE FROM images WHERE id = :image_id")
try:
result = self.session.execute(query, {"image_id": image_id})
self.session.commit()
return result.rowcount > 0
except Exception as e:
self.session.rollback()
raise Exception(f"Erreur lors de la suppression de l'image : {e}")
def close_connection(self):
"""Close the connection."""
self.session.close()
def get_image_count(self):
"""
Get the number of images in the `images` table.
Returns:
int: Number of images.
"""
query = text("SELECT COUNT(*) FROM images")
try:
result = self.session.execute(query).fetchone()
return result[0]
except Exception as e:
raise Exception(f"An error occurred while getting the image count: {e}") |