Source code for ace.database.utils

import logging
from ace.database.connection import ace_connection
from ace.database.query_builder import SelectQueryBuilder
from ace.database.query_builder import Condition
import pyodbc


[docs] class DatabaseUtils: """ A utility class for database operations. """
[docs] @staticmethod def get_database_connection(): """ Returns a database connection object. """ # Placeholder for actual database connection logic return "Database Connection"
[docs] class CodeUtils: """ A utility class for code-related operations. """
[docs] @staticmethod def fetchone(table, columns: list[str], conditions: list[Condition], sql_query=None): db_connection = ace_connection() cursor = db_connection.cursor if not sql_query: query_builder = SelectQueryBuilder(table_name=table, columns=columns) sql_query = query_builder.build_query(conditions=conditions) logging.debug(f"Executing query: {sql_query}") cursor.execute(sql_query) row = cursor.fetchone() if not row: logging.error("No data found for the given conditions.") db_connection.close() return None code_to_cache = dict(zip(columns, row)) logging.debug(f"Retrieved data: {code_to_cache}") db_connection.close() return code_to_cache
[docs] @staticmethod def fetchall(query, params=None): pass