lilfade/natureinpots_community
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

bunch of changes

Browse Source
This commit is contained in:
Bryson Shepard
2025-06-04 23:24:16 -05:00
parent 98c868113c
commit 6cf2fdec61
41 changed files with 1580 additions and 286 deletions
Download Patch File Download Diff File Expand all files Collapse all files

75
app/__init__.py
View File

@ -1,8 +1,9 @@
# app/__init__.py
import os
import json
import importlib
import glob
import importlib.util
from flask import Flask
from flask_sqlalchemy import SQLAlchemy
from flask_migrate import Migrate
@ -10,32 +11,52 @@ from flask_login import LoginManager
from flask_wtf.csrf import CSRFProtect
from dotenv import load_dotenv
# Load environment variables
load_dotenv()
# Initialize extensions
db = SQLAlchemy()
migrate = Migrate()
# ----------------------------------------------------------------
# 1) Initialize core extensions
# ----------------------------------------------------------------
db = SQLAlchemy()
migrate = Migrate()
login_manager = LoginManager()
csrf = CSRFProtect()
csrf = CSRFProtect()
def create_app():
app = Flask(__name__)
app.config.from_object('app.config.Config')
# Initialize core extensions
# Initialize extensions with app
csrf.init_app(app)
db.init_app(app)
migrate.init_app(app, db)
login_manager.init_app(app)
login_manager.login_view = 'auth.login'
# Register error handlers
# ----------------------------------------------------------------
# 2) Register error handlers
# ----------------------------------------------------------------
from .errors import bp as errors_bp
app.register_blueprint(errors_bp)
# Auto-discover and register plugins
# ----------------------------------------------------------------
# 3) Auto-load each plugin’s models.py so that SQLAlchemy metadata
# knows about every table (Plant, PlantOwnershipLog, PlantUpdate, etc.)
# ----------------------------------------------------------------
plugin_model_paths = glob.glob(os.path.join(os.path.dirname(__file__), '..', 'plugins', '*', 'models.py'))
for path in plugin_model_paths:
module_name = path.replace("/", ".").replace(".py", "")
try:
spec = importlib.util.spec_from_file_location(module_name, path)
mod = importlib.util.module_from_spec(spec)
spec.loader.exec_module(mod)
print(f"✅ (Startup) Loaded: {module_name}")
except Exception as e:
print(f"❌ (Startup) Failed to load {module_name}: {e}")
# ----------------------------------------------------------------
# 4) Auto-discover & register each plugin’s routes.py and CLI
# ----------------------------------------------------------------
plugin_path = os.path.abspath(os.path.join(os.path.dirname(__file__), '..', 'plugins'))
for plugin in os.listdir(plugin_path):
if plugin.endswith('.noload'):
@ -46,47 +67,37 @@ def create_app():
if not os.path.isdir(plugin_dir):
continue
# 1. Register routes
# --- (a) Register routes blueprint if present ---
route_file = os.path.join(plugin_dir, 'routes.py')
if os.path.isfile(route_file):
try:
spec = importlib.util.spec_from_file_location(f"plugins.{plugin}.routes", route_file)
mod = importlib.util.module_from_spec(spec)
mod = importlib.util.module_from_spec(spec)
spec.loader.exec_module(mod)
if hasattr(mod, 'bp'):
app.register_blueprint(mod.bp, strict_slashes=False)
print(f"✔️ Registered routes for plugin '{plugin}'")
except Exception as e:
print(f"[⚠️] Failed to load routes from plugin '{plugin}': {e}")
print(f"❌ Failed to load routes from plugin '{plugin}': {e}")
# Define paths
init_file = os.path.join(plugin_dir, '__init__.py')
# --- (b) Register CLI & entry point if present ---
init_file = os.path.join(plugin_dir, '__init__.py')
plugin_json = os.path.join(plugin_dir, 'plugin.json')
model_file = os.path.join(plugin_dir, 'models.py')
# 2. Register CLI commands and run entry point
if os.path.isfile(init_file):
try:
cli_module = importlib.import_module(f"plugins.{plugin}")
if hasattr(cli_module, 'register_cli'):
cli_module.register_cli(app)
print(f"✔️ Registered CLI for plugin '{plugin}'")
if os.path.isfile(plugin_json):
with open(plugin_json, 'r') as f:
meta = json.load(f)
entry = meta.get('entry_point')
if entry and hasattr(cli_module, entry):
getattr(cli_module, entry)(app)
print(f"✔️ Ran entry point '{entry}' for plugin '{plugin}'")
except Exception as e:
print(f"[⚠️] Failed to load CLI for plugin '{plugin}': {e}")
## 3. Auto-load plugin models for migrations
#if os.path.isfile(model_file):
# try:
# spec = importlib.util.spec_from_file_location(f"plugins.{plugin}.models", model_file)
# mod = importlib.util.module_from_spec(spec)
# spec.loader.exec_module(mod)
# except Exception as e:
# print(f"[⚠️] Failed to load models from plugin '{plugin}': {e}")
print(f"❌ Failed to load CLI for plugin '{plugin}': {e}")
@app.context_processor
def inject_current_year():
@ -94,9 +105,3 @@ def create_app():
return {'current_year': datetime.now().year}
return app
@login_manager.user_loader
def load_user(user_id):
from plugins.auth.models import User
return User.query.get(int(user_id))

111
app/neo4j_utils.py
View File

@ -1,32 +1,103 @@
# app/neo4j_utils.py
from neo4j import GraphDatabase
from flask import current_app
class Neo4jHandler:
def __init__(self, uri, user, password):
self.driver = GraphDatabase.driver(uri, auth=(user, password))
def __init__(self, uri=None, user=None, password=None):
# We read from current_app.config if nothing is passed in explicitly.
# If you already set NEO4J_URI / NEO4J_USER / NEO4J_PASSWORD in your config.py,
# these defaults will be overridden by those values automatically.
uri = uri or current_app.config.get("NEO4J_URI", "bolt://nip_neo4j:7687")
user = user or current_app.config.get("NEO4J_USER", "neo4j")
pw = password or current_app.config.get("NEO4J_PASSWORD", "your_password_here")
self.driver = GraphDatabase.driver(uri, auth=(user, pw))
def close(self):
self.driver.close()
def create_plant_node(self, uuid, name):
with self.driver.session() as session:
session.run(
"MERGE (p:Plant {uuid: $uuid}) "
"SET p.name = $name",
uuid=uuid, name=name
)
def create_plant_node(self, uuid: str, name: str = "Unknown"):
"""
MERGE a Plant node by UUID. On create, set its name.
We strip() and strip('"') in case the CSV had extra quotes or spaces around the UUID.
"""
if not uuid:
print("[⚠️] Skipped node creation: missing UUID")
return
def create_lineage(self, child_uuid, parent_uuid):
with self.driver.session() as session:
session.run(
"MATCH (child:Plant {uuid: $child_uuid}), (parent:Plant {uuid: $parent_uuid}) "
"MERGE (parent)-[:PARENT_OF]->(child)",
child_uuid=child_uuid, parent_uuid=parent_uuid
)
# Remove surrounding quotes or whitespace
uuid_clean = uuid.strip().strip('"')
name_clean = (name or "Unknown").strip()
def get_neo4j_handler():
uri = current_app.config['NEO4J_URI']
user = current_app.config['NEO4J_USER']
password = current_app.config['NEO4J_PASSWORD']
print(f"[ℹ️] (Neo4j) MERGE Plant node → uuid='{uuid_clean}', name='{name_clean}'")
try:
with self.driver.session() as session:
session.run(
"""
MERGE (p:Plant {uuid: $uuid})
ON CREATE SET p.name = $name
""",
uuid=uuid_clean,
name=name_clean
)
except Exception as e:
print(f"[❌] Neo4j node creation failed for UUID={uuid_clean}: {e}")
def create_lineage(self, child_uuid: str, parent_uuid: str):
"""
MATCH both child and parent by UUID, then MERGE a LINEAGE relationship.
Again, strip() any extraneous quotes or whitespace.
"""
if not child_uuid or not parent_uuid:
print(f"[⚠️] Skipped lineage creation: missing UUID(s) ({child_uuid!r} → {parent_uuid!r})")
return
child_clean = child_uuid.strip().strip('"')
parent_clean = parent_uuid.strip().strip('"')
print(f"[ℹ️] (Neo4j) Attempting to MERGE LINEAGE → child='{child_clean}', parent='{parent_clean}'")
try:
with self.driver.session() as session:
result = session.run(
"""
MATCH (c:Plant {uuid: $child_uuid})
MATCH (p:Plant {uuid: $parent_uuid})
MERGE (c)-[r:LINEAGE]->(p)
RETURN type(r) AS rel_type
""",
child_uuid=child_clean,
parent_uuid=parent_clean
)
record = result.single()
if record and record.get("rel_type") == "LINEAGE":
print(f"[✅] (Neo4j) Created LINEAGE → {child_clean} → {parent_clean}")
else:
print(f"[⚠️] (Neo4j) No LINEAGE created (nodes may not match) → {child_clean} → {parent_clean}")
except Exception as e:
print(f"[❌] Neo4j lineage creation failed: {e}")
def debug_check_node(self, uuid: str):
"""
Utility: check whether a Plant node with this UUID exists in Neo4j.
"""
uuid_clean = uuid.strip().strip('"')
with self.driver.session() as session:
result = session.run(
"MATCH (p:Plant {uuid: $uuid}) RETURN p",
uuid=uuid_clean
)
record = result.single()
if record:
print(f"[✅] (Neo4j) Node '{uuid_clean}' exists.")
else:
print(f"[❌] (Neo4j) Node '{uuid_clean}' NOT found.")
def get_neo4j_handler() -> Neo4jHandler:
"""
Factory: read NEO4J_URI / NEO4J_USER / NEO4J_PASSWORD from current_app.config.
"""
uri = current_app.config.get("NEO4J_URI", "bolt://nip_neo4j:7687")
user = current_app.config.get("NEO4J_USER", "neo4j")
password = current_app.config.get("NEO4J_PASSWORD", "your_password_here")
return Neo4jHandler(uri, user, password)