7 changed files with 120 additions and 11 deletions
@ -0,0 +1,98 @@ |
|||||
|
import sqlite3 |
||||
|
import threading |
||||
|
import json |
||||
|
|
||||
|
class DBController: |
||||
|
def __init__(self, db_path=':memory:', max_connections=10): |
||||
|
self.db_path = db_path |
||||
|
self.lock = threading.Lock() |
||||
|
self.conn = sqlite3.connect(db_path, check_same_thread=False) |
||||
|
self.conn.row_factory = sqlite3.Row |
||||
|
self.cursor = self.conn.cursor() |
||||
|
self.init_database() |
||||
|
|
||||
|
def init_database(self): |
||||
|
disk_conn = sqlite3.connect('pokemon_forms.db') |
||||
|
disk_cursor = disk_conn.cursor() |
||||
|
|
||||
|
# Create tables in the file-based database |
||||
|
self.create_pokemon_forms_table(disk_cursor) |
||||
|
|
||||
|
# Commit changes to the file-based database |
||||
|
disk_conn.commit() |
||||
|
|
||||
|
# Copy the file-based database to the in-memory database |
||||
|
disk_conn.backup(self.conn) |
||||
|
|
||||
|
# Close the file-based database connection |
||||
|
disk_conn.close() |
||||
|
|
||||
|
def save_changes(self): |
||||
|
with self.lock: |
||||
|
# Count the number of records before backup for verification |
||||
|
self.cursor.execute('SELECT COUNT(*) FROM pokemon_forms') |
||||
|
count = self.cursor.fetchone()[0] |
||||
|
print(f"Records in memory before backup: {count}") |
||||
|
|
||||
|
# Back up the master connection to disk |
||||
|
disk_conn = sqlite3.connect('pokemon_forms.db') |
||||
|
with disk_conn: |
||||
|
self.conn.backup(disk_conn) |
||||
|
disk_conn.close() |
||||
|
|
||||
|
def close(self): |
||||
|
self.save_changes() |
||||
|
self.conn.close() |
||||
|
|
||||
|
def create_pokemon_forms_table(self, cursor): |
||||
|
cursor.execute(''' |
||||
|
CREATE TABLE IF NOT EXISTS pokemon_forms ( |
||||
|
PFIC TEXT PRIMARY KEY, |
||||
|
data JSON NOT NULL |
||||
|
) |
||||
|
''') |
||||
|
|
||||
|
def add_pokemon_form(self, pfic, name, form_name, national_dex, generation, sprite_url): |
||||
|
data = { |
||||
|
"name": name, |
||||
|
"form_name": form_name, |
||||
|
"national_dex": national_dex, |
||||
|
"generation": generation, |
||||
|
"sprite_url": sprite_url, |
||||
|
"is_baby_form": False, |
||||
|
"storable_in_home": True, # Example additional field |
||||
|
} |
||||
|
|
||||
|
with self.lock: |
||||
|
self.cursor.execute(''' |
||||
|
INSERT OR REPLACE INTO pokemon_forms (PFIC, data) VALUES (?, ?) |
||||
|
''', (pfic, json.dumps(data))) |
||||
|
self.conn.commit() |
||||
|
print(f"Added: {pfic}, {name}") |
||||
|
|
||||
|
def get_pokemon_details(self, pfic): |
||||
|
self.cursor.execute(''' |
||||
|
SELECT JSON_EXTRACT(data, '$.name') AS name, |
||||
|
JSON_EXTRACT(data, '$.form_name') AS form_name, |
||||
|
JSON_EXTRACT(data, '$.national_dex') AS national_dex, |
||||
|
JSON_EXTRACT(data, '$.generation') AS generation, |
||||
|
JSON_EXTRACT(data, '$.is_baby_form') AS is_baby_form, |
||||
|
FROM pokemon_forms |
||||
|
WHERE PFIC = ? |
||||
|
''', (pfic,)) |
||||
|
results = self.cursor.fetchone() |
||||
|
return dict(results) |
||||
|
|
||||
|
def get_list_of_pokemon_forms(self): |
||||
|
self.cursor.execute(''' |
||||
|
SELECT JSON_EXTRACT(data, '$.name') AS name, |
||||
|
JSON_EXTRACT(data, '$.form_name') AS form_name, |
||||
|
JSON_EXTRACT(data, '$.national_dex') AS national_dex, |
||||
|
JSON_EXTRACT(data, '$.generation') AS generation, |
||||
|
JSON_EXTRACT(data, '$.is_baby_form') AS is_baby_form, |
||||
|
PFIC as pfic |
||||
|
FROM pokemon_forms |
||||
|
''',) |
||||
|
results = self.cursor.fetchall() |
||||
|
|
||||
|
return [dict(row) for row in results] |
||||
@ -0,0 +1,2 @@ |
|||||
|
from database.db_controller import DBController |
||||
|
db = DBController() |
||||
Loading…
Reference in new issue