Skip to content

XMRigDatabase

A class for handling database operations related to the XMRig miner.

Attributes:

Name Type Description
_engines Dict[str, Engine]

A dictionary to store database engines.
Source code in xmrig/db.py 
 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
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
class XMRigDatabase:
    """
    A class for handling database operations related to the XMRig miner.

    Attributes:
        _engines (Dict[str, Engine]): A dictionary to store database engines.
    """

    _engines = {}

    @classmethod
    def init_db(cls, db_url: str) -> Engine:
        """
        Initializes the database engine, if it already exists, it returns the existing engine.

        Args:
            db_url (str): Database URL for creating the engine.

        Returns:
            Engine: SQLAlchemy engine instance.

        Raises:
            XMRigDatabaseError: If an error occurs while initializing the database.
        """
        try:
            if db_url not in cls._engines:
                cls._engines[db_url] = create_engine(db_url)
            return cls._engines[db_url]
        except Exception as e:
            raise XMRigDatabaseError(e, traceback.print_exc(), f"An error occurred initializing the database:") from e

    # TODO: Implement across the codebase
    @classmethod
    def get_db(cls, db_url: str) -> Engine:
        """
        Returns the database engine for the specified database URL.

        Args:
            db_url (str): Database URL for creating the engine.

        Returns:
            Engine: SQLAlchemy engine instance.

        Raises:
            XMRigDatabaseError: If the database engine does not exist.
        """
        try:
            return cls._engines[db_url]
        except KeyError as e:
            raise XMRigDatabaseError(e, traceback.print_exc(), f"Database engine for '{db_url}' does not exist. Please initialize the database first.") from e

    @classmethod
    def check_table_exists(cls, db_url: str, table_name: str) -> bool:
        """
        Checks if the table exists in the database.

        Args:
            db_url (str): Database URL for creating the engine.
            table_name (str): Name of the table to check.

        Returns:
            bool: True if the table exists, False otherwise.

        Raises:
            XMRigDatabaseError: If an error occurs while checking if the table exists.
        """
        try:
            # Create an engine
            engine = cls.get_db(db_url)
            # Create an inspector
            inspector = inspect(engine)
            # Check if the table exists
            for i in inspector.get_table_names():
                if table_name[1:-1] in i:       # Remove the quotes from the table name
                    return True
            return False
        except Exception as e:
            raise XMRigDatabaseError(e, traceback.print_exc(), f"An error occurred checking if the table exists:") from e

    @classmethod
    def insert_data_to_db(cls, json_data: Dict[str, Any], table_name: str, db_url: str) -> None:
        """
        Inserts JSON data into the specified database table.

        Args:
            json_data (Dict[str, Any]): JSON data to insert.
            table_name (str): Name of the table to insert data into.
            db_url (str): Database URL for creating the engine.

        Raises:
            XMRigDatabaseError: If an error occurs while inserting data into the database.
        """
        try:
            # Create a dataframe with the required columns and data
            data = {
                "timestamp": [datetime.now()],
                "full_json": [json.dumps(json_data)]
            }
            engine = cls.get_db(db_url)
            df = pd.DataFrame(data)
            # Insert data into the database
            df.to_sql(table_name, engine, if_exists="append", index=False)
            log.debug("Data inserted successfully")
        except Exception as e:
            raise XMRigDatabaseError(e, traceback.print_exc(), f"An error occurred inserting data to the database:") from e

    @classmethod
    def fallback_to_db(cls, table_name: Union[str, List[str]], keys: List[Union[str, int]], db_url: str) -> Any:
        """
        Retrieves the data from the database using the provided table name.

        Args:
            table_name (Union[str, List[str]]): The name of the table or list of table names to use to retrieve the data.
            keys (List[Union[str, int]]): The keys to use to retrieve the data.
            db_url (str): The Database URL for creating the engine.

        Returns:
            Any: The retrieved data, or "N/A" if not available.

        Raises:
            XMRigDatabaseError: If an error occurs while retrieving data from the database.
        """
        column_name = "full_json"
        engine = cls.get_db(db_url)
        try:
            with engine.connect() as connection:
                # special handling for backends property, enables support for xmrig-mo fork
                if len(keys) < 1 and "backend" in table_name:
                    # get all backend tables and construct the response
                    backends = []
                    miner_name = table_name.split("-")[0].lstrip("'")
                    # Connect to the database and fetch the data in column_name from the table_name for each backend
                    if cls.check_table_exists(db_url, f"{miner_name}-cpu-backend"):
                        backends.append(json.loads(connection.execute(text(f"SELECT {column_name} FROM '{miner_name}-cpu-backend' ORDER BY timestamp DESC LIMIT 1")).fetchone()[0]))
                    if cls.check_table_exists(db_url, f"{miner_name}-opencl-backend"):
                        backends.append(json.loads(connection.execute(text(f"SELECT {column_name} FROM '{miner_name}-opencl-backend' ORDER BY timestamp DESC LIMIT 1")).fetchone()[0]))
                    if cls.check_table_exists(db_url, f"{miner_name}-cuda-backend"):
                        backends.append(json.loads(connection.execute(text(f"SELECT {column_name} FROM '{miner_name}-cuda-backend' ORDER BY timestamp DESC LIMIT 1")).fetchone()[0]))
                    return backends
                # default handling
                else:
                    # Connect to the database and fetch the data in column_name from the table_name
                    result = connection.execute(text(f"SELECT {column_name} FROM {table_name} ORDER BY timestamp DESC LIMIT 1"))
                    # Fetch the last item from the result
                    data = result.fetchone()
                    if data:
                        data = json.loads(data[0])
                    # if the first key is an int then that means we are dealing with the properties that require the
                    # backends tables, remove the first item from the keys list because the backends are stored in 
                    # individual tables
                    if isinstance(keys[0], int):
                        keys.pop(0)
                    # Use the list of keys/indices to access the correct data
                    if len(keys) > 0:
                        for key in keys:
                            data = data[key]
                    return data
            return "N/A"
        except Exception as e:
            raise XMRigDatabaseError(e, traceback.print_exc(), f"An error occurred retrieving data from the database:") from e
        finally:
            connection.close()

    @classmethod
    def delete_all_miner_data_from_db(cls, miner_name: str, db_url: str) -> None:
        """
        Deletes all tables related to a specific miner from the database.

        Args:
            miner_name (str): The unique name of the miner.
            db_url (str): Database URL for creating the engine.

        Raises:
            XMRigDatabaseError: If an error occurs while deleting the miner data from the database.
        """
        try:
            # Use quotes to avoid SQL syntax errors
            backends_tables = [f"'{miner_name}-cpu-backend'", f"'{miner_name}-opencl-backend'", f"'{miner_name}-cuda-backend'"]
            config_table = f"'{miner_name}-config'"
            summary_table = f"'{miner_name}-summary'"
            engine = cls.get_db(db_url)
            with engine.connect() as connection:
                # Wrap the raw SQL strings in SQLAlchemy's `text` function so it isn't a raw string
                connection.execute(text(f"DROP TABLE IF EXISTS {backends_tables[0]}"))
                connection.execute(text(f"DROP TABLE IF EXISTS {backends_tables[1]}"))
                connection.execute(text(f"DROP TABLE IF EXISTS {backends_tables[2]}"))
                connection.execute(text(f"DROP TABLE IF EXISTS {config_table}"))
                connection.execute(text(f"DROP TABLE IF EXISTS {summary_table}"))
            log.debug(f"All tables for '{miner_name}' have been deleted from the database")
        except Exception as e:
            raise XMRigDatabaseError(e, traceback.print_exc(), f"An error occurred deleting miner '{miner_name}' from the database:") from e

check_table_exists(db_url, table_name) classmethod

Checks if the table exists in the database.

Parameters:

Name Type Description Default
db_url str

Database URL for creating the engine.

 required
table_name str

Name of the table to check.

 required

Returns:

Name Type Description
bool bool

True if the table exists, False otherwise.

Raises:

Type Description
XMRigDatabaseError

If an error occurs while checking if the table exists.
Source code in xmrig/db.py 
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
@classmethod
def check_table_exists(cls, db_url: str, table_name: str) -> bool:
    """
    Checks if the table exists in the database.

    Args:
        db_url (str): Database URL for creating the engine.
        table_name (str): Name of the table to check.

    Returns:
        bool: True if the table exists, False otherwise.

    Raises:
        XMRigDatabaseError: If an error occurs while checking if the table exists.
    """
    try:
        # Create an engine
        engine = cls.get_db(db_url)
        # Create an inspector
        inspector = inspect(engine)
        # Check if the table exists
        for i in inspector.get_table_names():
            if table_name[1:-1] in i:       # Remove the quotes from the table name
                return True
        return False
    except Exception as e:
        raise XMRigDatabaseError(e, traceback.print_exc(), f"An error occurred checking if the table exists:") from e

delete_all_miner_data_from_db(miner_name, db_url) classmethod

Deletes all tables related to a specific miner from the database.

Parameters:

Name Type Description Default
miner_name str

The unique name of the miner.

 required
db_url str

Database URL for creating the engine.

 required

Raises:

Type Description
XMRigDatabaseError

If an error occurs while deleting the miner data from the database.
Source code in xmrig/db.py 
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
@classmethod
def delete_all_miner_data_from_db(cls, miner_name: str, db_url: str) -> None:
    """
    Deletes all tables related to a specific miner from the database.

    Args:
        miner_name (str): The unique name of the miner.
        db_url (str): Database URL for creating the engine.

    Raises:
        XMRigDatabaseError: If an error occurs while deleting the miner data from the database.
    """
    try:
        # Use quotes to avoid SQL syntax errors
        backends_tables = [f"'{miner_name}-cpu-backend'", f"'{miner_name}-opencl-backend'", f"'{miner_name}-cuda-backend'"]
        config_table = f"'{miner_name}-config'"
        summary_table = f"'{miner_name}-summary'"
        engine = cls.get_db(db_url)
        with engine.connect() as connection:
            # Wrap the raw SQL strings in SQLAlchemy's `text` function so it isn't a raw string
            connection.execute(text(f"DROP TABLE IF EXISTS {backends_tables[0]}"))
            connection.execute(text(f"DROP TABLE IF EXISTS {backends_tables[1]}"))
            connection.execute(text(f"DROP TABLE IF EXISTS {backends_tables[2]}"))
            connection.execute(text(f"DROP TABLE IF EXISTS {config_table}"))
            connection.execute(text(f"DROP TABLE IF EXISTS {summary_table}"))
        log.debug(f"All tables for '{miner_name}' have been deleted from the database")
    except Exception as e:
        raise XMRigDatabaseError(e, traceback.print_exc(), f"An error occurred deleting miner '{miner_name}' from the database:") from e

fallback_to_db(table_name, keys, db_url) classmethod

Retrieves the data from the database using the provided table name.

Parameters:

Name Type Description Default
table_name Union[str, List[str]]

The name of the table or list of table names to use to retrieve the data.

 required
keys List[Union[str, int]]

The keys to use to retrieve the data.

 required
db_url str

The Database URL for creating the engine.

 required

Returns:

Name Type Description
Any Any

The retrieved data, or "N/A" if not available.

Raises:

Type Description
XMRigDatabaseError

If an error occurs while retrieving data from the database.
Source code in xmrig/db.py 
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
@classmethod
def fallback_to_db(cls, table_name: Union[str, List[str]], keys: List[Union[str, int]], db_url: str) -> Any:
    """
    Retrieves the data from the database using the provided table name.

    Args:
        table_name (Union[str, List[str]]): The name of the table or list of table names to use to retrieve the data.
        keys (List[Union[str, int]]): The keys to use to retrieve the data.
        db_url (str): The Database URL for creating the engine.

    Returns:
        Any: The retrieved data, or "N/A" if not available.

    Raises:
        XMRigDatabaseError: If an error occurs while retrieving data from the database.
    """
    column_name = "full_json"
    engine = cls.get_db(db_url)
    try:
        with engine.connect() as connection:
            # special handling for backends property, enables support for xmrig-mo fork
            if len(keys) < 1 and "backend" in table_name:
                # get all backend tables and construct the response
                backends = []
                miner_name = table_name.split("-")[0].lstrip("'")
                # Connect to the database and fetch the data in column_name from the table_name for each backend
                if cls.check_table_exists(db_url, f"{miner_name}-cpu-backend"):
                    backends.append(json.loads(connection.execute(text(f"SELECT {column_name} FROM '{miner_name}-cpu-backend' ORDER BY timestamp DESC LIMIT 1")).fetchone()[0]))
                if cls.check_table_exists(db_url, f"{miner_name}-opencl-backend"):
                    backends.append(json.loads(connection.execute(text(f"SELECT {column_name} FROM '{miner_name}-opencl-backend' ORDER BY timestamp DESC LIMIT 1")).fetchone()[0]))
                if cls.check_table_exists(db_url, f"{miner_name}-cuda-backend"):
                    backends.append(json.loads(connection.execute(text(f"SELECT {column_name} FROM '{miner_name}-cuda-backend' ORDER BY timestamp DESC LIMIT 1")).fetchone()[0]))
                return backends
            # default handling
            else:
                # Connect to the database and fetch the data in column_name from the table_name
                result = connection.execute(text(f"SELECT {column_name} FROM {table_name} ORDER BY timestamp DESC LIMIT 1"))
                # Fetch the last item from the result
                data = result.fetchone()
                if data:
                    data = json.loads(data[0])
                # if the first key is an int then that means we are dealing with the properties that require the
                # backends tables, remove the first item from the keys list because the backends are stored in 
                # individual tables
                if isinstance(keys[0], int):
                    keys.pop(0)
                # Use the list of keys/indices to access the correct data
                if len(keys) > 0:
                    for key in keys:
                        data = data[key]
                return data
        return "N/A"
    except Exception as e:
        raise XMRigDatabaseError(e, traceback.print_exc(), f"An error occurred retrieving data from the database:") from e
    finally:
        connection.close()

get_db(db_url) classmethod

Returns the database engine for the specified database URL.

Parameters:

Name Type Description Default
db_url str

Database URL for creating the engine.

 required

Returns:

Name Type Description
Engine Engine

SQLAlchemy engine instance.

Raises:

Type Description
XMRigDatabaseError

If the database engine does not exist.
Source code in xmrig/db.py 
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
@classmethod
def get_db(cls, db_url: str) -> Engine:
    """
    Returns the database engine for the specified database URL.

    Args:
        db_url (str): Database URL for creating the engine.

    Returns:
        Engine: SQLAlchemy engine instance.

    Raises:
        XMRigDatabaseError: If the database engine does not exist.
    """
    try:
        return cls._engines[db_url]
    except KeyError as e:
        raise XMRigDatabaseError(e, traceback.print_exc(), f"Database engine for '{db_url}' does not exist. Please initialize the database first.") from e

init_db(db_url) classmethod

Initializes the database engine, if it already exists, it returns the existing engine.

Parameters:

Name Type Description Default
db_url str

Database URL for creating the engine.

 required

Returns:

Name Type Description
Engine Engine

SQLAlchemy engine instance.

Raises:

Type Description
XMRigDatabaseError

If an error occurs while initializing the database.
Source code in xmrig/db.py 
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
@classmethod
def init_db(cls, db_url: str) -> Engine:
    """
    Initializes the database engine, if it already exists, it returns the existing engine.

    Args:
        db_url (str): Database URL for creating the engine.

    Returns:
        Engine: SQLAlchemy engine instance.

    Raises:
        XMRigDatabaseError: If an error occurs while initializing the database.
    """
    try:
        if db_url not in cls._engines:
            cls._engines[db_url] = create_engine(db_url)
        return cls._engines[db_url]
    except Exception as e:
        raise XMRigDatabaseError(e, traceback.print_exc(), f"An error occurred initializing the database:") from e

insert_data_to_db(json_data, table_name, db_url) classmethod

Inserts JSON data into the specified database table.

Parameters:

Name Type Description Default
json_data Dict[str, Any]

JSON data to insert.

 required
table_name str

Name of the table to insert data into.

 required
db_url str

Database URL for creating the engine.

 required

Raises:

Type Description
XMRigDatabaseError

If an error occurs while inserting data into the database.
Source code in xmrig/db.py 
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
@classmethod
def insert_data_to_db(cls, json_data: Dict[str, Any], table_name: str, db_url: str) -> None:
    """
    Inserts JSON data into the specified database table.

    Args:
        json_data (Dict[str, Any]): JSON data to insert.
        table_name (str): Name of the table to insert data into.
        db_url (str): Database URL for creating the engine.

    Raises:
        XMRigDatabaseError: If an error occurs while inserting data into the database.
    """
    try:
        # Create a dataframe with the required columns and data
        data = {
            "timestamp": [datetime.now()],
            "full_json": [json.dumps(json_data)]
        }
        engine = cls.get_db(db_url)
        df = pd.DataFrame(data)
        # Insert data into the database
        df.to_sql(table_name, engine, if_exists="append", index=False)
        log.debug("Data inserted successfully")
    except Exception as e:
        raise XMRigDatabaseError(e, traceback.print_exc(), f"An error occurred inserting data to the database:") from e