91 lines
3.5 KiB
Python
91 lines
3.5 KiB
Python
from odoorpc import ODOO
|
|
import os
|
|
import sys
|
|
from services.config import Config
|
|
|
|
|
|
class OdooConnection:
|
|
def __init__(self, config_path="config/settings.yaml", instance_name=None):
|
|
"""Initialize with a config file and optional instance name."""
|
|
self.config = Config(config_path)
|
|
self.instance_name = instance_name
|
|
self.instances = self._load_instances()
|
|
self.connections = {} # Store active Odoo connections
|
|
|
|
def _load_instances(self):
|
|
"""Load Odoo instances from the config."""
|
|
instances = self.config.get_instances()
|
|
if not instances:
|
|
raise ValueError("No Odoo instances found in the configuration.")
|
|
|
|
if self.instance_name:
|
|
# Filter for the specified instance
|
|
filtered = [i for i in instances if i["name"] == self.instance_name]
|
|
if not filtered:
|
|
raise ValueError(
|
|
f"Instance '{self.instance_name}' not found in configuration."
|
|
)
|
|
return filtered
|
|
return instances
|
|
|
|
def connect(self, instance_name=None):
|
|
"""Connect to a specific instance or all instances if none specified."""
|
|
target_instances = (
|
|
[i for i in self.instances if i["name"] == instance_name]
|
|
if instance_name
|
|
else self.instances
|
|
)
|
|
|
|
for instance in target_instances:
|
|
if instance["name"] in self.connections:
|
|
print(f"Using existing connection for {instance['name']}")
|
|
continue
|
|
|
|
try:
|
|
# Create Odoo RPC connection
|
|
odoo = ODOO(instance["host"], port=instance["port"])
|
|
odoo.login(
|
|
instance["database"], instance["username"], instance["password"]
|
|
)
|
|
self.connections[instance["name"]] = odoo
|
|
print(
|
|
f"Connected to {instance['host']}:{instance['port']} - {instance['database']}"
|
|
)
|
|
except Exception as e:
|
|
print(f"Failed to connect to {instance['name']}: {e}")
|
|
raise
|
|
|
|
def get_connection(self, instance_name):
|
|
"""Get an existing connection for an instance, connecting if necessary."""
|
|
if instance_name not in self.connections:
|
|
self.connect(instance_name)
|
|
return self.connections.get(instance_name)
|
|
|
|
def disconnect(self, instance_name=None):
|
|
"""Disconnect from a specific instance or all instances."""
|
|
target_instances = [instance_name] if instance_name else self.connections.keys()
|
|
|
|
for name in target_instances:
|
|
if name in self.connections:
|
|
# odoorpc doesn't have an explicit disconnect, so we just remove the reference
|
|
del self.connections[name]
|
|
print(f"Disconnected from instance: {name}")
|
|
|
|
def get_instances(self):
|
|
"""Return the list of configured instances."""
|
|
return self.instances
|
|
|
|
def execute(self, instance_name, model, method, *args, **kwargs):
|
|
"""Execute a method on a model for a specific instance."""
|
|
odoo = self.get_connection(instance_name)
|
|
if not odoo:
|
|
raise ValueError(f"No connection available for instance '{instance_name}'")
|
|
|
|
try:
|
|
model_obj = odoo.env[model]
|
|
result = getattr(model_obj, method)(*args, **kwargs)
|
|
return result
|
|
except Exception as e:
|
|
print(f"Error executing {method} on {model} for {instance_name}: {e}")
|
|
raise
|