Spaces:
Runtime error
Runtime error
| import logging | |
| from db.repository import Repository, get_db_conn | |
| # Setup logging (configure as needed) | |
| logging.basicConfig(level=logging.INFO) | |
| class GetDatabase(Repository): | |
| async def execute_query(self, query, params=None, fetch_one=False): | |
| """ | |
| Helper function to execute SQL queries and handle exceptions. | |
| """ | |
| try: | |
| print(fetch_one) | |
| if fetch_one: | |
| results = await self._fetch_one(query, params) | |
| print(results) | |
| else: | |
| results = await self.get_by_query(query, params) | |
| print("result execute query : ", results) | |
| return results if results else None | |
| except Exception as e: | |
| logging.error(f"An error occurred while executing query: {e}") | |
| return None | |
| async def get_data(self, title): | |
| """ | |
| Fetch the first result matching the given title from the metadata table. | |
| """ | |
| query = """ | |
| SELECT * FROM Metadata | |
| WHERE title = %s | |
| limit 5; | |
| """ | |
| try: | |
| results = await self.execute_query(query, (title,), fetch_one=True) | |
| return results | |
| except Exception as e: | |
| logging.error(f"An error occurred while get data: {e}") | |
| return None | |
| async def get_all_data(self): | |
| """ | |
| Fetch all data from the metadata table. | |
| """ | |
| query = """ | |
| SELECT * FROM Metadata | |
| """ | |
| results = await self.execute_query(query) | |
| return results | |
| async def get_data_by_id(self, id): | |
| query = f""" | |
| SELECT * FROM Metadata WHERE id = :id | |
| """ | |
| param = {"id" : id} | |
| try: | |
| results = await self.execute_query(query, param) | |
| print('Query successful, results: %s', results) | |
| return results[0] if results else None | |
| except Exception as e: | |
| print('Error fetching data by ID %s: %s', id, e) | |
| return None |