Py-KMS-Organization/py-kms
1
0
Fork 0
mirror of https://github.com/Py-KMS-Organization/py-kms.git synced 2026-04-21 06:03:43 +00:00
Code Issues Projects Releases Packages Wiki Activity

Migrated to named columns to prevent any accidential re-order

Browse source 
Signed-off-by: simonmicro <simon@simonmicro.de>
This commit is contained in:
simonmicro 2025-12-06 21:25:05 +01:00
parent 6c57a8e5b4
commit 1c28865f3d
No known key found for this signature in database
GPG key ID: 033A4D4CE4E063D6
1 changed files with 14 additions and 22 deletions
Download patch file Download diff file Expand all files Collapse all files

36
py-kms/pykms_Sql.py
View file

@ -7,17 +7,7 @@ import logging
#-------------------------------------------------------------------------------------------------------------------------------------------------------- #--------------------------------------------------------------------------------------------------------------------------------------------------------
loggersrv = logging.getLogger('logsrv') loggersrv = logging.getLogger('logsrv')
_column_name_to_index = { _column_names = ('clientMachineId', 'machineName', 'applicationId', 'skuId', 'licenseStatus', 'lastRequestTime', 'kmsEpid', 'requestCount', 'lastRequestIP')
'clientMachineId': 0,
'machineName': 1,
'applicationId': 2,
'skuId': 3,
'licenseStatus': 4,
'lastRequestTime': 5,
'kmsEpid': 6,
'requestCount': 7,
'lastRequestIP': 8,
}
# sqlite3 is optional. # sqlite3 is optional.
available = False available = False
@ -68,17 +58,18 @@ def sql_get_all(dbName):
if not os.path.isfile(dbName): if not os.path.isfile(dbName):
return None return None
with sqlite3.connect(dbName) as con: with sqlite3.connect(dbName) as con:
con.row_factory = sqlite3.Row
cur = con.cursor() cur = con.cursor()
cur.execute(f"SELECT {', '.join(_column_name_to_index.keys())} FROM clients") cur.execute(f"SELECT {', '.join(_column_names)} FROM clients")
clients = [] clients = []
for row in cur.fetchall(): for row in cur.fetchall():
loggersrv.debug(f"Row: {row}") loggersrv.debug(f"Row: {row}")
obj = {} obj = {}
for col_name, index in _column_name_to_index.items(): for col_name in _column_names:
if col_name == "lastRequestTime": if col_name == "lastRequestTime":
obj[col_name] = datetime.fromtimestamp(row[_column_name_to_index['lastRequestTime']]).isoformat() obj[col_name] = datetime.fromtimestamp(row['lastRequestTime']).isoformat()
else: else:
obj[col_name] = row[index] obj[col_name] = row[col_name]
loggersrv.debug(f"Obj: {obj}") loggersrv.debug(f"Obj: {obj}")
clients.append(obj) clients.append(obj)
return clients return clients
@ -88,36 +79,37 @@ def sql_update(dbName, infoDict):
return return
# make sure all column names are present # make sure all column names are present
for col_name in _column_name_to_index.keys(): for col_name in _column_names:
if col_name in ["requestCount", "kmsEpid"]: if col_name in ["requestCount", "kmsEpid"]:
continue continue
if col_name not in infoDict: if col_name not in infoDict:
raise ValueError(f"infoDict is missing required column: {col_name}") raise ValueError(f"infoDict is missing required column: {col_name}")
with sqlite3.connect(dbName) as con: with sqlite3.connect(dbName) as con:
con.row_factory = sqlite3.Row
cur = con.cursor() cur = con.cursor()
cur.execute(f"SELECT {', '.join(_column_name_to_index.keys())} FROM clients WHERE clientMachineId=:clientMachineId AND applicationId=:applicationId;", infoDict) cur.execute(f"SELECT {', '.join(_column_names)} FROM clients WHERE clientMachineId=:clientMachineId AND applicationId=:applicationId;", infoDict)
data = cur.fetchone() data = cur.fetchone()
if not data: if not data:
# Insert new row with all given info # Insert new row with all given info
infoDict["kmsEpid"] = "" # Default empty value infoDict["kmsEpid"] = "" # Default empty value
infoDict["requestCount"] = 1 infoDict["requestCount"] = 1
cur.execute(f"""INSERT INTO clients ({', '.join(_column_name_to_index.keys())}) cur.execute(f"""INSERT INTO clients ({', '.join(_column_names)})
VALUES ({', '.join(':' + col for col in _column_name_to_index.keys())});""", infoDict) VALUES ({', '.join(':' + col for col in _column_names)});""", infoDict)
else: else:
# Update only changed columns # Update only changed columns
common_postfix = "WHERE clientMachineId=:clientMachineId AND applicationId=:applicationId" common_postfix = "WHERE clientMachineId=:clientMachineId AND applicationId=:applicationId"
def update_column_if_changed(column_name, new_value): def update_column_if_changed(column_name, new_value):
assert "clientMachineId" in infoDict and "applicationId" in infoDict, "infoDict must contain 'clientMachineId' and 'applicationId'" assert "clientMachineId" in infoDict and "applicationId" in infoDict, "infoDict must contain 'clientMachineId' and 'applicationId'"
if column_name not in _column_name_to_index: if column_name not in _column_names:
raise ValueError(f"Unknown column name: {column_name}") raise ValueError(f"Unknown column name: {column_name}")
if data[_column_name_to_index[column_name]] != new_value: if data[column_name] != new_value:
query = f"UPDATE clients SET {column_name}=:value {common_postfix}" query = f"UPDATE clients SET {column_name}=:value {common_postfix}"
cur.execute(query, {"value": new_value, "clientMachineId": infoDict['clientMachineId'], "applicationId": infoDict['applicationId']}) cur.execute(query, {"value": new_value, "clientMachineId": infoDict['clientMachineId'], "applicationId": infoDict['applicationId']})
# Dynamically check and maybe update all columns # Dynamically check and maybe update all columns
for column_name in _column_name_to_index.keys(): for column_name in _column_names:
if column_name in ["clientMachineId", "applicationId", "requestCount"]: if column_name in ["clientMachineId", "applicationId", "requestCount"]:
continue # Skip these columns continue # Skip these columns
if column_name == "kmsEpid": if column_name == "kmsEpid":