Sync Streamline
This commit is contained in:
parent
330bf13465
commit
787a81641f
14 changed files with 405 additions and 398 deletions
1
.gitignore
vendored
1
.gitignore
vendored
|
|
@ -1,3 +1,4 @@
|
|||
.dev
|
||||
prototype_client.py
|
||||
.idea
|
||||
.venv
|
||||
|
|
|
|||
1
core/.metadata.md
Normal file
1
core/.metadata.md
Normal file
|
|
@ -0,0 +1 @@
|
|||
this is june 13 version
|
||||
|
|
@ -10,8 +10,7 @@ class Constants:
|
|||
"TICKET_API_BASE_URL", "https://ticket.hydraveil.net"
|
||||
)
|
||||
|
||||
SP_API_BASE_URL: Final[str] = os.environ.get('SP_API_BASE_URL', 'https://dev-us-002.simplifiedprivacy.net/api/v1')
|
||||
# SP_API_BASE_URL: Final[str] = os.environ.get('SP_API_BASE_URL', 'https://api.hydraveil.net/api/v1')
|
||||
SP_API_BASE_URL: Final[str] = os.environ.get('SP_API_BASE_URL', 'https://api.hydraveil.net/api/v1')
|
||||
PING_URL: Final[str] = os.environ.get('PING_URL', 'https://api.hydraveil.net/api/v1/health')
|
||||
|
||||
CONNECTION_RETRY_INTERVAL: Final[int] = int(os.environ.get('CONNECTION_RETRY_INTERVAL', '5'))
|
||||
|
|
|
|||
|
|
@ -1,5 +1,7 @@
|
|||
# new sync refactor:
|
||||
from core.services.sync.sync_service import coordinate_cache_sync
|
||||
|
||||
from core.errors.logger import logger
|
||||
# prior versions:
|
||||
from core.Constants import Constants
|
||||
from core.Errors import UnknownClientPathError, UnknownClientVersionError, CommandNotFoundError
|
||||
from core.controllers.ApplicationController import ApplicationController
|
||||
|
|
@ -47,37 +49,39 @@ class ClientController:
|
|||
def sync(client_observer: ClientObserver = None, connection_observer: ConnectionObserver = None):
|
||||
|
||||
from core.models.manage.session_management import init_session, close_session
|
||||
init_session()
|
||||
print("I sync'ed")
|
||||
|
||||
# Use Cached Method:
|
||||
changed_tables = coordinate_cache_sync(ClientObserver, ConnectionObserver)
|
||||
init_session()
|
||||
result = coordinate_cache_sync(ClientObserver, ConnectionObserver)
|
||||
|
||||
# SPOOFING
|
||||
# changed_tables = ['version', 'applications', 'application_versions', 'client_version', 'operators', 'locations', 'subscriptions']
|
||||
logger.info(f"We got a Result from the API of {result}")
|
||||
|
||||
print(f"changed_tables: {changed_tables}")
|
||||
close_session()
|
||||
|
||||
# this should be same or any errors:
|
||||
if isinstance(changed_tables, dict):
|
||||
# check result is "same"?
|
||||
# or an errro
|
||||
# Outright Error:
|
||||
if not result["success"]:
|
||||
error_msg = result["error"]
|
||||
if client_observer is not None:
|
||||
client_observer.notify('synchronizing', f'Error! {error_msg}')
|
||||
return
|
||||
|
||||
# Same:
|
||||
changed_tables = result["changed_tables"]
|
||||
if not changed_tables:
|
||||
if client_observer is not None:
|
||||
client_observer.notify('synchronized')
|
||||
return
|
||||
|
||||
# now it's likely to have results:
|
||||
if isinstance(changed_tables, list):
|
||||
quantity_of_changes = len(changed_tables)
|
||||
# We only make it past this point if there's New Data
|
||||
|
||||
print(f"quantity_of_changes is {quantity_of_changes}")
|
||||
|
||||
# sanity check:
|
||||
if quantity_of_changes >= 0:
|
||||
from core.controllers.ConnectionController import ConnectionController
|
||||
ConnectionController.with_preferred_connection(task=ClientController.__sync, changed_tables=changed_tables, client_observer=client_observer, connection_observer=connection_observer)
|
||||
else:
|
||||
if client_observer is not None:
|
||||
client_observer.notify('synchronized')
|
||||
# Fetch and update those models...
|
||||
try:
|
||||
from core.controllers.ConnectionController import ConnectionController
|
||||
ConnectionController.with_preferred_connection(task=ClientController.__sync, changed_tables=changed_tables, client_observer=client_observer, connection_observer=connection_observer)
|
||||
except:
|
||||
# sync failed here,
|
||||
if client_observer is not None:
|
||||
client_observer.notify('synchronizing', 'Error! Sync Failed.')
|
||||
|
||||
@staticmethod
|
||||
def update(client_observer: ClientObserver = None, connection_observer: ConnectionObserver = None):
|
||||
|
|
@ -93,49 +97,58 @@ class ClientController:
|
|||
|
||||
return path
|
||||
|
||||
# def cached_sync(client_observer: Optional[ClientObserver] = None, proxies: Optional[dict] = None):
|
||||
|
||||
@staticmethod
|
||||
def __sync(changed_tables: list, client_observer: Optional[ClientObserver] = None, proxies: Optional[dict] = None):
|
||||
|
||||
# SPOOFING
|
||||
# changed_tables = ['version', 'applications', 'application_versions', 'client_version', 'operators', 'locations', 'subscriptions']
|
||||
|
||||
if client_observer is not None:
|
||||
client_observer.notify('synchronizing')
|
||||
|
||||
if "applications" in changed_tables:
|
||||
print("sync applications")
|
||||
logger.info("Sync applications..")
|
||||
if client_observer is not None:
|
||||
client_observer.notify('synchronizing', 'Fetching Browser List')
|
||||
# noinspection PyProtectedMember
|
||||
ApplicationController._sync(proxies=proxies)
|
||||
|
||||
if "application_versions" in changed_tables:
|
||||
print("sync application_versions")
|
||||
logger.info("Sync Application Versions..")
|
||||
if client_observer is not None:
|
||||
client_observer.notify('synchronizing', 'Fetching Browser Version List')
|
||||
# noinspection PyProtectedMember
|
||||
ApplicationVersionController._sync(proxies=proxies)
|
||||
|
||||
if "client_version" in changed_tables:
|
||||
print("sync client_version")
|
||||
logger.info("Sync of client version")
|
||||
if client_observer is not None:
|
||||
client_observer.notify('synchronizing', 'Fetching Client Version List')
|
||||
# noinspection PyProtectedMember
|
||||
ClientVersionController._sync(proxies=proxies)
|
||||
|
||||
if "operators" in changed_tables:
|
||||
print("sync operators")
|
||||
logger.info("Sync of Operators")
|
||||
if client_observer is not None:
|
||||
client_observer.notify('synchronizing', 'Fetching Operators List')
|
||||
# noinspection PyProtectedMember
|
||||
OperatorController._sync(proxies=proxies)
|
||||
|
||||
if "locations" in changed_tables:
|
||||
print("sync locations")
|
||||
logger.info("Sync of Locations")
|
||||
if client_observer is not None:
|
||||
client_observer.notify('synchronizing', 'Fetching Locations List')
|
||||
# noinspection PyProtectedMember
|
||||
LocationController._sync(proxies=proxies)
|
||||
|
||||
if "subscriptions" in changed_tables:
|
||||
print("sync subscriptions")
|
||||
logger.info("Sync of Subscriptions")
|
||||
if client_observer is not None:
|
||||
client_observer.notify('synchronizing', 'Fetching Subscription List')
|
||||
# noinspection PyProtectedMember
|
||||
SubscriptionPlanController._sync(proxies=proxies)
|
||||
|
||||
ConfigurationController.update_last_synced_at()
|
||||
|
||||
logger.info("Entire Sync Completed Successfully")
|
||||
if client_observer is not None:
|
||||
client_observer.notify('synchronized')
|
||||
|
||||
|
|
|
|||
|
|
@ -4,7 +4,7 @@ from sqlalchemy.exc import SQLAlchemyError
|
|||
|
||||
def get_from_model(model_class: Type, **filters) -> List[Dict[str, Any]]:
|
||||
"""
|
||||
Retrieve data from a model.
|
||||
Retrieve data from a model. Keeping this generic for reuse
|
||||
|
||||
Args:
|
||||
model_class: The ORM model class
|
||||
|
|
|
|||
|
|
@ -1,14 +1,15 @@
|
|||
from core.errors.logger import logger
|
||||
from sqlalchemy.orm import sessionmaker
|
||||
from sqlalchemy import create_engine
|
||||
from typing import Type, Dict, Any
|
||||
from sqlalchemy.exc import IntegrityError, SQLAlchemyError
|
||||
from core.models.python_models.Base import Base
|
||||
from core.models.python_models.CachedSync import CachedSync
|
||||
from core.models.orm_models.Base import Base
|
||||
from core.models.orm_models.CachedSync import CachedSync
|
||||
from core.models.manage.session_management import get_session
|
||||
|
||||
def insert_into_model(model_class: Type, all_data: dict | list, override=False) -> bool:
|
||||
"""
|
||||
Generic ORM insert for any model.
|
||||
Generic ORM insert for any model. Keeping it generic for reuse.
|
||||
|
||||
Args:
|
||||
model_class: The ORM model class
|
||||
|
|
@ -19,15 +20,16 @@ def insert_into_model(model_class: Type, all_data: dict | list, override=False)
|
|||
True if successful
|
||||
"""
|
||||
|
||||
print(f"all_data is {all_data}")
|
||||
logger.info(f"All the public data preparing to be inserted is {all_data}")
|
||||
|
||||
with get_session() as session:
|
||||
# Step 1: Wipe the table if override=True
|
||||
if override:
|
||||
logger.info(f"First, WIPING the pre-existing data for {model_class.__name__}. Are you sure you intended to completely delete the old data?")
|
||||
session.query(model_class).delete()
|
||||
session.commit()
|
||||
|
||||
print(f"Starting insert for {model_class.__name__}")
|
||||
logger.info(f"Starting insert for {model_class.__name__}")
|
||||
|
||||
# Normalize to list for uniform handling
|
||||
data_list = all_data if isinstance(all_data, list) else [all_data]
|
||||
|
|
@ -38,6 +40,7 @@ def insert_into_model(model_class: Type, all_data: dict | list, override=False)
|
|||
session.add(instance)
|
||||
|
||||
session.commit()
|
||||
logger.info(f"Completed! SQL Insertion Successful following the {model_class.__name__} Model’s Type Rules")
|
||||
return True
|
||||
|
||||
except TypeError as e:
|
||||
|
|
@ -56,83 +59,3 @@ def insert_into_model(model_class: Type, all_data: dict | list, override=False)
|
|||
print(f"Generic exception caught: {e}")
|
||||
session.rollback()
|
||||
raise e
|
||||
|
||||
|
||||
|
||||
# def insert_into_model(model_class: Type, all_data: list, override=False) -> Any:
|
||||
# """
|
||||
# Generic ORM insert for any model. But not used for the legacy codebase yet.
|
||||
|
||||
# Args:
|
||||
# model_class: The ORM model class
|
||||
# json_data: Dictionary with data matching the model
|
||||
|
||||
# Returns:
|
||||
# The created instance
|
||||
# """
|
||||
|
||||
# print(f"all_data is {all_data}")
|
||||
|
||||
# with get_session() as session:
|
||||
# # Step 1: Wipe the table if override=True
|
||||
# if override:
|
||||
# session.query(model_class).delete()
|
||||
# session.commit()
|
||||
|
||||
# print("we made it past getting the session")
|
||||
|
||||
# print(f"Starting insert for {model_class.__name__}")
|
||||
|
||||
# type_of_data = type(all_data)
|
||||
# print(f"all_data type is: {type_of_data}")
|
||||
|
||||
# if isinstance(all_data, dict):
|
||||
# try:
|
||||
# instance = model_class(**all_data)
|
||||
# session.add(instance)
|
||||
# session.commit()
|
||||
# except TypeError as e:
|
||||
# print(f"TypeError caught: {e}")
|
||||
# session.rollback()
|
||||
# raise ValueError(f"Invalid fields for {model_class.__name__}: {e}")
|
||||
# except IntegrityError as e:
|
||||
# print(f"IntegrityError caught: {e.orig}")
|
||||
# session.rollback()
|
||||
# raise ValueError(f"Constraint violation: {e.orig}")
|
||||
# except SQLAlchemyError as e:
|
||||
# print(f"SQLAlchemyError caught: {e}")
|
||||
# session.rollback()
|
||||
# raise
|
||||
# except Exception as e:
|
||||
# print(f"Generic exception caught: {e}")
|
||||
# session.rollback()
|
||||
# raise e
|
||||
# return True
|
||||
|
||||
# elif isinstance(all_data, list):
|
||||
# for each_json in all_data:
|
||||
# try:
|
||||
# instance = model_class(**each_json)
|
||||
# session.add(instance)
|
||||
# session.commit()
|
||||
# except TypeError as e:
|
||||
# print(f"TypeError caught: {e}")
|
||||
# session.rollback()
|
||||
# raise ValueError(f"Invalid fields for {model_class.__name__}: {e}")
|
||||
# except IntegrityError as e:
|
||||
# print(f"IntegrityError caught: {e.orig}")
|
||||
# session.rollback()
|
||||
# raise ValueError(f"Constraint violation: {e.orig}")
|
||||
# except SQLAlchemyError as e:
|
||||
# print(f"SQLAlchemyError caught: {e}")
|
||||
# session.rollback()
|
||||
# raise
|
||||
# except Exception as e:
|
||||
# print(f"Generic exception caught: {e}")
|
||||
# session.rollback()
|
||||
# raise e
|
||||
# return True
|
||||
|
||||
# else:
|
||||
# print("invalid data type")
|
||||
# return False
|
||||
|
|
@ -1,33 +0,0 @@
|
|||
# db.py
|
||||
from core.models.python_models.Base import Base
|
||||
from sqlalchemy import create_engine
|
||||
from sqlalchemy.orm import scoped_session, sessionmaker
|
||||
from sqlalchemy.exc import SQLAlchemyError
|
||||
|
||||
# Create engine
|
||||
try:
|
||||
engine = create_engine('sqlite:///data.db', echo=False)
|
||||
except Exception as e:
|
||||
print(f"Failed to create engine: {e}")
|
||||
raise
|
||||
|
||||
Session = scoped_session(sessionmaker(bind=engine))
|
||||
|
||||
def init_session():
|
||||
"""Initialize database tables."""
|
||||
try:
|
||||
Base.metadata.create_all(engine)
|
||||
except SQLAlchemyError as e:
|
||||
print(f"Failed to initialize database: {e}")
|
||||
raise
|
||||
|
||||
def get_session():
|
||||
"""Get the thread-local session."""
|
||||
return Session()
|
||||
|
||||
def close_session():
|
||||
"""Clean up the thread-local session."""
|
||||
try:
|
||||
Session.remove()
|
||||
except Exception as e:
|
||||
print(f"Error closing session: {e}")
|
||||
|
|
@ -1,14 +1,22 @@
|
|||
from sqlalchemy.orm import sessionmaker
|
||||
from sqlalchemy import create_engine
|
||||
from core.models.python_models.Base import Base
|
||||
from core.models.orm_models.Base import Base
|
||||
from core.Constants import Constants
|
||||
|
||||
engine = create_engine("sqlite:///data.db")
|
||||
database_path = Constants.HV_STORAGE_DATABASE_PATH
|
||||
engine = create_engine(f"sqlite:///{database_path}")
|
||||
Session = sessionmaker(bind=engine)
|
||||
|
||||
_session = None
|
||||
|
||||
def init_session():
|
||||
"""Call this once at application startup"""
|
||||
"""
|
||||
Called once at application startup using global variables for the session,
|
||||
which initialize prior to being called with a false flag.
|
||||
|
||||
The reason for this is so the app can load prior to be called,
|
||||
while keeping it a globally accessible variable
|
||||
"""
|
||||
global _session
|
||||
Base.metadata.create_all(engine)
|
||||
_session = Session()
|
||||
|
|
@ -20,7 +28,12 @@ def get_session():
|
|||
return _session
|
||||
|
||||
def close_session():
|
||||
"""Call this when shutting down the application"""
|
||||
"""
|
||||
Used prior to shutting down the application to break SQL connections.
|
||||
|
||||
In the current refactor, it may be used prior to shut down, such as sync completion,
|
||||
because a low amount of the total models currently use the new system.
|
||||
"""
|
||||
global _session
|
||||
if _session:
|
||||
_session.close()
|
||||
|
|
|
|||
|
|
@ -1,19 +0,0 @@
|
|||
from contextlib import contextmanager
|
||||
from sqlalchemy.orm import sessionmaker
|
||||
from sqlalchemy import create_engine
|
||||
|
||||
|
||||
engine = create_engine("sqlite:///alchemy.db")
|
||||
Session = sessionmaker(bind=engine)
|
||||
|
||||
Base.metadata.create_all(engine)
|
||||
|
||||
@contextmanager
|
||||
def get_session():
|
||||
session = Session()
|
||||
try:
|
||||
yield session
|
||||
finally:
|
||||
session.close()
|
||||
|
||||
|
||||
9
core/models/orm_models/Base.py
Normal file
9
core/models/orm_models/Base.py
Normal file
|
|
@ -0,0 +1,9 @@
|
|||
# base.py
|
||||
from sqlalchemy.orm import declarative_base
|
||||
|
||||
Base = declarative_base()
|
||||
|
||||
"""
|
||||
This will eventually replace the legacy base model,
|
||||
but is still required, even if empty, to setup the declarative base.
|
||||
"""
|
||||
|
|
@ -2,12 +2,18 @@ from sqlalchemy import Integer, String, ForeignKey
|
|||
from sqlalchemy.orm import declarative_base, mapped_column, Mapped, relationship
|
||||
from typing import Optional
|
||||
from sqlalchemy.orm import Mapped
|
||||
from core.models.python_models.Base import Base
|
||||
from core.models.orm_models.Base import Base
|
||||
|
||||
"""
|
||||
This is one of the SQLAlchemy Models from the refactor.
|
||||
|
||||
That’s why it lacks functions to get or insert data.
|
||||
"""
|
||||
|
||||
class CachedSync(Base):
|
||||
__tablename__ = 'cached_sync'
|
||||
|
||||
# version of the cached sync itself primary key
|
||||
# version of the cached sync itself is the primary key
|
||||
version: Mapped[int] = mapped_column(Integer, primary_key=True)
|
||||
|
||||
applications: Mapped[Optional[str]] = mapped_column(Integer, nullable=True, default=None)
|
||||
|
|
@ -16,33 +22,3 @@ class CachedSync(Base):
|
|||
operators: Mapped[Optional[str]] = mapped_column(Integer, nullable=True, default=None)
|
||||
locations: Mapped[Optional[str]] = mapped_column(Integer, nullable=True, default=None)
|
||||
subscriptions: Mapped[Optional[str]] = mapped_column(Integer, nullable=True, default=None)
|
||||
|
||||
# to use:
|
||||
# Lookup by composite primary key
|
||||
# record = session.get(Hello, (b_value))
|
||||
|
||||
|
||||
|
||||
# # from core.models.Model import Model
|
||||
# from dataclasses import dataclass
|
||||
# from typing import Union, Optional
|
||||
|
||||
# _table_name: str = 'cached_sync'
|
||||
|
||||
# _table_definition: str = """
|
||||
# 'version' int UNIQUE,
|
||||
# 'applications' int,
|
||||
# 'application_versions' int,
|
||||
# 'operators' int,
|
||||
# 'locations' int,
|
||||
# 'subscriptions' int,
|
||||
# """
|
||||
|
||||
# @dataclass
|
||||
# class CachedSync(Model):
|
||||
# version: int
|
||||
# applications: int
|
||||
# application_versions: int
|
||||
# operators: int
|
||||
# locations: int
|
||||
# subscriptions: int
|
||||
|
|
@ -1,4 +0,0 @@
|
|||
# base.py (or models/base.py)
|
||||
from sqlalchemy.orm import declarative_base
|
||||
|
||||
Base = declarative_base()
|
||||
|
|
@ -1,69 +1,70 @@
|
|||
|
||||
def compare_tables(new_data: dict, old_table: dict) -> tuple:
|
||||
import logging
|
||||
from typing import Optional
|
||||
|
||||
type_of_new = type(new_data)
|
||||
type_of_old = type(old_table)
|
||||
print(type_of_new)
|
||||
print(type_of_old)
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
# ============== FILTER ==================
|
||||
new_is_dictionary = isinstance(new_data, dict)
|
||||
old_is_dictionary = isinstance(old_table, dict)
|
||||
if new_is_dictionary and old_is_dictionary:
|
||||
print("Filter Passed. Both things are dictionaries.")
|
||||
else:
|
||||
return None, None
|
||||
|
||||
# ============== PREPARE ==================
|
||||
|
||||
# what are the tables?
|
||||
new_keys = new_data.keys()
|
||||
old_keys = old_table.keys()
|
||||
|
||||
print(f"new_keys is {new_keys}")
|
||||
print(f"old_keys is {old_keys}")
|
||||
|
||||
def compare_tables(new_data: dict, old_table: dict) -> tuple[list[str], list[str]]:
|
||||
"""
|
||||
Compare version numbers in new and old metadata dictionaries.
|
||||
|
||||
Args:
|
||||
new_data: Latest metadata from API (e.g., {"version": 2, "model_one": 2})
|
||||
old_table: Previously cached metadata from database
|
||||
|
||||
Returns:
|
||||
Tuple of (changed_tables, new_model_types)
|
||||
- changed_tables: model keys where version increased
|
||||
- new_model_types: model keys that don't exist in old_table (reserved for future use)
|
||||
|
||||
Raises:
|
||||
TypeError: If either argument is not a dictionary
|
||||
ValueError: If a model version is not an integer
|
||||
"""
|
||||
# Validate inputs
|
||||
if not isinstance(new_data, dict) or not isinstance(old_table, dict):
|
||||
raise TypeError("Both new_data and old_table must be dictionaries")
|
||||
|
||||
changed_tables = []
|
||||
new_data_types = []
|
||||
|
||||
# ============== EVALUATE ==================
|
||||
|
||||
for new_key in new_keys:
|
||||
print(f"This iteration's new_key is {new_key}")
|
||||
each_value = new_data[new_key]
|
||||
print(f"This value is {each_value}")
|
||||
|
||||
# is the VALUE we're evaluating even a number?
|
||||
if not isinstance(each_value, int):
|
||||
print("skipping, it's not an int")
|
||||
new_model_types = []
|
||||
|
||||
for model_key, new_version in new_data.items():
|
||||
# Skip the "version" metadata key itself
|
||||
if model_key == "version":
|
||||
continue
|
||||
|
||||
# do we even know about this new table category?
|
||||
if new_key not in old_keys:
|
||||
new_data_types.append(new_key)
|
||||
|
||||
# Type check: versions must be integers
|
||||
if not isinstance(new_version, int):
|
||||
raise ValueError(
|
||||
f"Model '{model_key}' has non-integer version: {new_version!r}. "
|
||||
f"Expected int."
|
||||
)
|
||||
|
||||
# Identify new model types (not in old data)
|
||||
if model_key not in old_table:
|
||||
new_model_types.append(model_key)
|
||||
continue
|
||||
|
||||
# Identify changed models (version increased)
|
||||
if new_version > old_table[model_key]:
|
||||
changed_tables.append(model_key)
|
||||
|
||||
return changed_tables, new_model_types
|
||||
|
||||
# we know about it, let's see if it's new:
|
||||
if each_value > old_table[new_key]:
|
||||
changed_tables.append(new_key)
|
||||
|
||||
return changed_tables, new_data_types
|
||||
|
||||
# ============== TESTING ==================
|
||||
# ISOLATION TEST:
|
||||
"""
|
||||
new_data = {
|
||||
"a": 1,
|
||||
"b": 2,
|
||||
"c": 4,
|
||||
"d": 1
|
||||
}
|
||||
old_data = {
|
||||
"a": 1,
|
||||
"b": 2,
|
||||
"c": 3,
|
||||
}
|
||||
changed_tables, new_data_types = compare_tables(new_data, old_data)
|
||||
print(f"changed tables: {changed_tables}")
|
||||
print(f"new data types: {new_data_types}")
|
||||
"""
|
||||
# ============== ISOLATION TESTING ==================
|
||||
# SPOOF ISOLATION TEST:
|
||||
# new_data = {
|
||||
# "a": 1,
|
||||
# "b": 2,
|
||||
# "c": 4,
|
||||
# "d": 1
|
||||
# }
|
||||
# old_data = {
|
||||
# "a": 1,
|
||||
# "b": 2,
|
||||
# "c": 3,
|
||||
# }
|
||||
# changed_tables, new_data_types = compare_tables(new_data, old_data)
|
||||
# print(f"changed tables: {changed_tables}")
|
||||
# print(f"new data types: {new_data_types}")
|
||||
|
|
|
|||
|
|
@ -12,7 +12,7 @@ from core.services.sync.compare_tables import compare_tables
|
|||
from core.models.manage.session_management import get_session
|
||||
from core.models.manage.get_from_model import get_from_model
|
||||
from core.models.manage.insert import insert_into_model
|
||||
from core.models.python_models.CachedSync import CachedSync
|
||||
from core.models.orm_models.CachedSync import CachedSync
|
||||
|
||||
from core.Constants import Constants
|
||||
from core.observers.BaseObserver import BaseObserver
|
||||
|
|
@ -24,161 +24,288 @@ from core.observers.ConnectionObserver import ConnectionObserver
|
|||
# generic
|
||||
from sqlalchemy import func
|
||||
from typing import Optional
|
||||
from typing import Optional
|
||||
|
||||
|
||||
def _get_cached_metadata():
|
||||
"""Retrieve cached metadata from the database.
|
||||
|
||||
Returns:
|
||||
tuple: (old_data dict, previous_highest version number)
|
||||
If no cache exists: ({}, 0)
|
||||
|
||||
Raises:
|
||||
Exception: Any database query errors propagate up.
|
||||
"""
|
||||
old_data_as_list = get_from_model(CachedSync)
|
||||
|
||||
if not old_data_as_list:
|
||||
logger.warning("No cached sync data in database. Treating as first sync.")
|
||||
return {}, 0
|
||||
|
||||
if not isinstance(old_data_as_list, list):
|
||||
logger.critical("Error with get_from_model improperly returning non-lists.")
|
||||
return {}, 0
|
||||
|
||||
old_data = old_data_as_list[0]
|
||||
previous_highest = old_data.get("version", 0)
|
||||
logger.debug(f"Cached metadata version: {previous_highest}")
|
||||
|
||||
return old_data, previous_highest
|
||||
|
||||
|
||||
def _get_changed_models(new_data, old_data):
|
||||
"""Determine which models changed between API and cached versions.
|
||||
|
||||
Args:
|
||||
new_data: Dict of new metadata from API
|
||||
old_data: Dict of old metadata from cache (empty dict if first sync)
|
||||
|
||||
Returns:
|
||||
tuple: (changed_tables list, new_model_types list)
|
||||
|
||||
Raises:
|
||||
TypeError: If inputs have invalid structure
|
||||
ValueError: If version values are invalid
|
||||
"""
|
||||
if not old_data:
|
||||
# First sync: no baseline, treat all models as needing fetch
|
||||
changed_tables = [key for key in new_data if key != "version"]
|
||||
new_model_types = []
|
||||
logger.info(f"First sync: fetching all models: {changed_tables}")
|
||||
else:
|
||||
# Subsequent sync: compare against baseline
|
||||
changed_tables, new_model_types = compare_tables(new_data, old_data)
|
||||
|
||||
return changed_tables, new_model_types
|
||||
|
||||
|
||||
def _filter_valid_models(data, invalid_keys):
|
||||
"""Remove invalid model types from data dict.
|
||||
|
||||
Args:
|
||||
data: Dict of metadata
|
||||
invalid_keys: List of keys to exclude
|
||||
|
||||
Returns:
|
||||
Dict with invalid keys removed
|
||||
"""
|
||||
filtered = {}
|
||||
for key, value in data.items():
|
||||
if key not in invalid_keys:
|
||||
filtered[key] = value
|
||||
return filtered
|
||||
|
||||
|
||||
def save_data(data: dict) -> bool:
|
||||
"""Save metadata to database."""
|
||||
try:
|
||||
logger.debug(f"Inserting metadata into database: {data}")
|
||||
did_it_work = insert_into_model(CachedSync, data, True)
|
||||
|
||||
if did_it_work:
|
||||
logger.debug("Metadata saved successfully")
|
||||
else:
|
||||
logger.warning("insert_into_model returned False")
|
||||
|
||||
return did_it_work
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Failed to save metadata to database: {str(e)}", exc_info=True)
|
||||
return False
|
||||
|
||||
|
||||
def _log_what_changed(new_model_types, changed_tables):
|
||||
if new_model_types:
|
||||
logger.debug(f"New model types detected (reserved): {new_model_types}")
|
||||
|
||||
if changed_tables:
|
||||
logger.info(f"Models to sync: {changed_tables}")
|
||||
else:
|
||||
logger.info("No models changed between versions.")
|
||||
|
||||
|
||||
# Highest Level in this Flow:
|
||||
def coordinate_cache_sync(
|
||||
client_observer: Optional[ClientObserver] = None,
|
||||
connection_observer: Optional[ConnectionObserver] = None
|
||||
) -> dict:
|
||||
"""
|
||||
Fetch version metadata from API and return changed/new model keys.
|
||||
|
||||
Returns:
|
||||
Dictionary with keys:
|
||||
- 'success' (bool): Whether comparison succeeded
|
||||
- 'changed_tables' (list): Model keys that changed
|
||||
- 'new_model_types' (list): New model keys (reserved for future use)
|
||||
- 'error' (str or None): Error message if success is False
|
||||
"""
|
||||
|
||||
# ==================== GET & VALIDATE DATA ============================
|
||||
api_result = _get_sync_cache_from_api(ClientObserver, ConnectionObserver)
|
||||
|
||||
# get data from the endpoint:
|
||||
new_data_payload = get_sync_cache_from_api(ClientObserver, ConnectionObserver)
|
||||
# we trust the '_get_sync_cache_from_api' function to give us a dictionary:
|
||||
if not api_result.get("success"):
|
||||
error_msg = api_result.get("error", "Unknown API error")
|
||||
logger.error(f"API sync failed: {error_msg}")
|
||||
return {
|
||||
"success": False,
|
||||
"changed_tables": [],
|
||||
"new_model_types": [],
|
||||
"error": error_msg,
|
||||
}
|
||||
|
||||
print("")
|
||||
print(f"We have new data: {new_data_payload}")
|
||||
print(type(new_data_payload))
|
||||
print("")
|
||||
# Extract the data from the client's own trusted function,
|
||||
new_data = api_result.get("data")
|
||||
|
||||
# is is a dictionary?
|
||||
if not isinstance(new_data_payload, dict):
|
||||
return {"valid": False, "error_msg": "invalid_format_reply"}
|
||||
|
||||
if "data" not in new_data_payload:
|
||||
return new_data_payload
|
||||
|
||||
new_data = new_data_payload.get("data", None)
|
||||
|
||||
# SPOOFING:
|
||||
# new_data = {
|
||||
# "version": 1,
|
||||
# "applications": 1,
|
||||
# "application_versions": 1,
|
||||
# "client_version": 1,
|
||||
# "operators": 1,
|
||||
# "locations": 1,
|
||||
# "subscriptions": 1
|
||||
# }
|
||||
# Validate the API's payload, (we don't trust the API's structure)
|
||||
if not isinstance(new_data, dict):
|
||||
error_msg = "API returned invalid metadata: 'data' is not a dict"
|
||||
logger.error(error_msg)
|
||||
return {
|
||||
"success": False,
|
||||
"changed_tables": [],
|
||||
"new_model_types": [],
|
||||
"error": error_msg,
|
||||
}
|
||||
|
||||
if "version" not in new_data:
|
||||
return new_data # it should have the error in there.
|
||||
error_msg = "API returned invalid metadata: missing 'version' key"
|
||||
logger.error(error_msg)
|
||||
return {
|
||||
"success": False,
|
||||
"changed_tables": [],
|
||||
"new_model_types": [],
|
||||
"error": error_msg,
|
||||
}
|
||||
|
||||
new_highest = new_data.get("version", None)
|
||||
|
||||
print(f"new highest version is {new_highest}")
|
||||
new_highest = new_data.get("version")
|
||||
logger.debug(f"API metadata version: {new_highest}")
|
||||
|
||||
if not new_highest: # version is empty
|
||||
return new_data # it should have the error in there.
|
||||
|
||||
# ==================== INITIAL SQL SESSION =====================
|
||||
session = get_session()
|
||||
|
||||
# Get the highest version already saved:
|
||||
previous_highest = session.query(func.max(CachedSync.version)).scalar()
|
||||
print(f"previous_highest is {previous_highest}")
|
||||
|
||||
# if this is the first sync, we have no data to compare it to,
|
||||
if previous_highest is None:
|
||||
print("No previous previous_highest or Previous data. Saving..")
|
||||
|
||||
# so save the data,
|
||||
saved = save_data(new_data)
|
||||
if saved:
|
||||
print("Saved!")
|
||||
|
||||
# then the "changed" keys are all of the new ones,
|
||||
changed_tables_keys = new_data.keys()
|
||||
changed_tables = list(changed_tables_keys)
|
||||
return changed_tables
|
||||
|
||||
# ==================== EVALUATE DATA =====================
|
||||
|
||||
print(f"previous_highest is {previous_highest}")
|
||||
print(f"new_highest is {new_highest}")
|
||||
|
||||
# bail on the rest of this function if it's the same:
|
||||
if previous_highest >= new_highest:
|
||||
return {"result": "same"}
|
||||
else:
|
||||
print("The data is new! We now evaluate...")
|
||||
|
||||
# ==================== COMPARE DATA ============================
|
||||
|
||||
# we are only doing this if we actually have old data to compare it to.
|
||||
query = {"version": previous_highest}
|
||||
old_data_as_list = get_from_model(CachedSync, **query)
|
||||
|
||||
# if it's a list, get it out of there,
|
||||
old_data = old_data_as_list[0]
|
||||
|
||||
print(f"old data is: {old_data}")
|
||||
|
||||
if old_data:
|
||||
print(f"We got the old data successfully. Now comparing..")
|
||||
|
||||
# let's see which tables are NEW:
|
||||
changed_tables, new_data_types = compare_tables(new_data, old_data)
|
||||
|
||||
print("")
|
||||
print(f"changed_tables is {changed_tables}")
|
||||
print(f"new_data_types is {new_data_types}")
|
||||
print("")
|
||||
|
||||
if new_data_types:
|
||||
quantity_of_new_types = len(new_data_types)
|
||||
else:
|
||||
# we can't fetch the OLD data due to database error, so we need to replace it all,
|
||||
print("database error. add logic here for quantity_of_new_types.")
|
||||
changed_tables_keys = new_data.keys()
|
||||
changed_tables = list(changed_tables_keys)
|
||||
quantity_of_new_types = len(changed_tables)
|
||||
|
||||
# ==================== FINALLY =====================
|
||||
return changed_tables
|
||||
|
||||
### Add error logging to this:
|
||||
def save_data(data: dict) -> bool:
|
||||
print(f"about to insert the NEW data into the database.. {data}")
|
||||
# Fetch old cached version from database
|
||||
try:
|
||||
did_it_work = insert_into_model(CachedSync, data, True)
|
||||
print(f"Did it insert? {did_it_work}")
|
||||
except:
|
||||
print("Error! We Failed to Insert the New Data!")
|
||||
did_it_work = False
|
||||
old_data, previous_highest = _get_cached_metadata()
|
||||
|
||||
except Exception as e:
|
||||
error_msg = f"Database query failed: {str(e)}"
|
||||
logger.error(error_msg, exc_info=True)
|
||||
return {
|
||||
"success": False,
|
||||
"changed_tables": [],
|
||||
"new_model_types": [],
|
||||
"error": error_msg,
|
||||
}
|
||||
|
||||
return did_it_work
|
||||
# Version hasn't changed
|
||||
if new_highest == previous_highest:
|
||||
logger.info(f"Versions match (both {new_highest}). No sync needed.")
|
||||
return {
|
||||
"success": True,
|
||||
"changed_tables": [],
|
||||
"new_model_types": [],
|
||||
"error": None,
|
||||
}
|
||||
|
||||
# Determine which models changed
|
||||
try:
|
||||
changed_tables, new_model_types = _get_changed_models(new_data, old_data)
|
||||
|
||||
except (TypeError, ValueError) as e:
|
||||
error_msg = f"Invalid metadata: {str(e)}"
|
||||
logger.error(error_msg)
|
||||
return {
|
||||
"success": False,
|
||||
"changed_tables": [],
|
||||
"new_model_types": [],
|
||||
"error": error_msg,
|
||||
}
|
||||
|
||||
# this is not saving it to SQL, it's a temp debug log:
|
||||
_log_what_changed(new_model_types, changed_tables)
|
||||
|
||||
# Filter out new model types for the upcoming save:
|
||||
filtered_data = _filter_valid_models(new_data, new_model_types)
|
||||
|
||||
# Save only currently existing models to SQL:
|
||||
save_successful = save_data(filtered_data)
|
||||
if save_successful:
|
||||
return {
|
||||
"success": True,
|
||||
"changed_tables": changed_tables,
|
||||
"new_model_types": new_model_types,
|
||||
"error": None,
|
||||
}
|
||||
else:
|
||||
return {
|
||||
"success": False,
|
||||
"changed_tables": [],
|
||||
"new_model_types": [],
|
||||
"error": "Failed to save metadata to database",
|
||||
}
|
||||
|
||||
|
||||
def get_sync_cache_from_api(
|
||||
|
||||
|
||||
def _get_sync_cache_from_api(
|
||||
client_observer: Optional[ClientObserver] = None,
|
||||
connection_observer: Optional[ConnectionObserver] = None
|
||||
) -> dict:
|
||||
logger.info("Syncing with the API...")
|
||||
|
||||
print("we are syncing with the API..")
|
||||
|
||||
# get the base url & verify it:
|
||||
# Get and validate the base URL
|
||||
rejected_list = [None, False, ""]
|
||||
# base_url = Constants.SP_API_BASE_URL
|
||||
base_url = 'https://dev-us-002.simplifiedprivacy.net/api/v1' # temp spoof
|
||||
base_url = Constants.SP_API_BASE_URL
|
||||
if base_url in rejected_list:
|
||||
# notification = "Base URL is Empty, so it can't fetch prices.."
|
||||
# client_observer.notify(notification)
|
||||
return {"valid": False, "error_code": "invalid_url"}
|
||||
error_msg = "Invalid base URL from configuration"
|
||||
logger.error(error_msg)
|
||||
return {"success": False, "data": None, "error": error_msg}
|
||||
|
||||
# combine with endpoint:
|
||||
# Construct the full endpoint URL
|
||||
url = f"{base_url}/cachedsync"
|
||||
logger.debug(f"API endpoint: {url}")
|
||||
|
||||
try:
|
||||
print("getting data from server...")
|
||||
logger.debug("Fetching data from API server...")
|
||||
sync_results = get_data_from_server(url, connection_observer)
|
||||
|
||||
if sync_results in rejected_list:
|
||||
return {"valid": False, "error_code": "sync_failed"}
|
||||
error_msg = "API returned no data"
|
||||
logger.error(error_msg)
|
||||
return {"success": False, "data": None, "error": error_msg}
|
||||
|
||||
logger.debug(f"Inside the sync controller, sync_results is: {sync_results}")
|
||||
except:
|
||||
return {"valid": False, "error_code": "sync_failed"}
|
||||
logger.debug(f"Raw API response: {sync_results}")
|
||||
|
||||
return sync_results
|
||||
final_result = sync_results.get("data", None)
|
||||
|
||||
if final_result:
|
||||
logger.info("Successfully retrieved sync versions metadata from API")
|
||||
return {"success": True, "data": final_result, "error": None}
|
||||
|
||||
error_msg = "API response missing 'data' field"
|
||||
logger.error(error_msg)
|
||||
return {"success": False, "data": None, "error": error_msg}
|
||||
|
||||
except Exception as e:
|
||||
error_msg = f"API fetch failed: {str(e)}"
|
||||
logger.error(error_msg)
|
||||
return {"success": False, "data": None, "error": error_msg}
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
# ============== ISOLATION TESTING ==================
|
||||
|
||||
# SPOOF ISOLATION TEST:
|
||||
# def _get_sync_cache_from_api(
|
||||
# final_result = {
|
||||
# "version": 5,
|
||||
# "applications": 3,
|
||||
# "application_versions": 2,
|
||||
# "client_version": 2,
|
||||
# "operators": 2,
|
||||
# "locations": 2,
|
||||
# "subscriptions": 1
|
||||
# }
|
||||
# return {"success": True, "data": final_result, "error": None}
|
||||
|
|
|
|||
Loading…
Reference in a new issue