feat: refactor Analysis class initialization and improve region support

Restructured Analysis class to separate initialization from execution flow using new init() method pattern. Modified TagoContext from TypedDict to class for better runtime flexibility. Updated documentation examples to reflect Python runtime instead of Deno.

Author: mateuscardosodeveloper

.vscode/settings.json

@@ -8,6 +8,7 @@
         "ignoretz",
         "PYPI",
         "pytest",
-        "serie_number"
+        "serie_number",
+        "Tago"
     ]
 }

src/tagoio_sdk/modules/Analysis/Analysis.py

@@ -44,50 +44,69 @@ def my_analysis(context, scope):
             # Your analysis logic here
             print("Processing data...")
 
-        analysis = Analysis(my_analysis, {"token": "your-analysis-token"})
+        Analysis.use(analysis=my_analysis, params={"token": "your-analysis-token"})
         ```
 
-    Example: Environment variables
+    Example: Analysis with EU region
         ```python
+        from tagoio_sdk import Analysis
+
         def my_analysis(context, scope):
-            env = context.environment
-            api_key = next((e["value"] for e in env if e["key"] == "API_KEY"), None)
+            context.log("Running in EU region")
+            print("Environment:", context.environment)
+
+        # Using Analysis.use() method
+        Analysis.use(analysis=my_analysis, params={"token": "your-analysis-token", "region": "eu-w1"})
+        ```
 
-        analysis = Analysis(my_analysis)
+    Example: Analysis with Tago Deploy
+        ```python
+        from tagoio_sdk import Analysis
+
+        def my_analysis(context, scope):
+            context.log("Running in TDeploy")
+            print("Scope:", scope)
+
+        # Tago Deploy requires a dictionary with tdeploy ID
+        Analysis.use(
+            analysis=my_analysis,
+            params={
+                "token": "your-analysis-token",
+                "region": {"tdeploy": "your-tdeploy-id"}
+            }
+        )
         ```
 
-    Example: Manual start control
+    Example: Environment variables
         ```python
-        analysis = Analysis(my_analysis, {
-            "token": "token",
-            "autostart": False
-        })
+        def my_analysis(context, scope):
+            env = context.environment
+            api_key = next((e["value"] for e in env if e["key"] == "API_KEY"), None)
 
-        # Start analysis manually
-        analysis.start()
+        Analysis.use(analysis=my_analysis, params={"token": "your-analysis-token"})
         ```
     """
 
-    def __init__(self, analysis: AnalysisFunction, params: Optional[AnalysisConstructorParams] = None):
+    def __init__(self, params: Optional[AnalysisConstructorParams] = None):
         if params is None:
             params = {"token": "unknown"}
 
         super().__init__(params)
         self.params = params
-        self.analysis = analysis
-        self.started = False
         self._running = True
 
-        if params.get("autostart", True):
-            self.start()
+    def init(self, analysis: AnalysisFunction):
+        self.analysis = analysis
 
-    def start(self) -> None:
-        if self.started:
-            return
+        if not os.environ.get("T_ANALYSIS_TOKEN") and self.params.get("token"):
+            os.environ["T_ANALYSIS_TOKEN"] = self.params.get("token")
 
-        self.started = True
+        # Configure runtime region
+        runtimeRegion = getRegionObj(self.params["region"]) if self.params.get("region") else None
+        if runtimeRegion:
+            setRuntimeRegion(runtimeRegion)
 
-        if not os.environ.get("T_ANALYSIS_CONTEXT"):
+        if T_ANALYSIS_CONTEXT is None:
             self._localRuntime()
         else:
             self._runOnTagoIO()
@@ -96,13 +115,13 @@ def _runOnTagoIO(self) -> None:
         if not self.analysis or not callable(self.analysis):
             raise TypeError("Invalid analysis function")
 
-        # Create context object
-        context = {
-            "log": print,
-            "token": os.environ.get("T_ANALYSIS_TOKEN", ""),
-            "environment": JSONParseSafe(os.environ.get("T_ANALYSIS_ENV", "[]"), []),
-            "analysis_id": os.environ.get("T_ANALYSIS_ID", ""),
-        }
+        def context():
+            pass
+
+        context.log = print
+        context.token = os.environ.get("T_ANALYSIS_TOKEN", "")
+        context.analysis_id = (JSONParseSafe(os.environ.get("T_ANALYSIS_ENV", "[]"), []),)
+        context.environment = (os.environ.get("T_ANALYSIS_ID", ""),)
 
         data = JSONParseSafe(os.environ.get("T_ANALYSIS_DATA", "[]"), [])
 
@@ -148,12 +167,13 @@ def log(*args: Any) -> None:
             except Exception as e:
                 print(f"Console error: {e}", file=sys.stderr)
 
-        context = {
-            "log": log,
-            "token": token,
-            "environment": environment,
-            "analysis_id": analysisID,
-        }
+        def context():
+            pass
+
+        context.log = log
+        context.token = token
+        context.environment = environment
+        context.analysis_id = analysisID
 
         # Execute analysis function
         if inspect.iscoroutinefunction(self.analysis):
@@ -180,7 +200,7 @@ def _localRuntime(self) -> None:
             analysis = None
 
         if not analysis:
-            print("¬ Error :: Analysis not found or not active.", file=sys.stderr)
+            print("¬ Error :: Analysis not found or not active or invalid analysis token.", file=sys.stderr)
             return
 
         if analysis.get("run_on") != "external":
@@ -220,7 +240,7 @@ def _localRuntime(self) -> None:
                         payload.get("environment", []),
                         payload.get("data", []),
                         payload.get("analysis_id", ""),
-                        payload.get("token", self.params.get("token", "")),
+                        self.token,
                     )
                 except Exception as e:
                     print(f"¬ Error processing event: {e}", file=sys.stderr)
@@ -256,12 +276,4 @@ def my_analysis(context, scope):
         if params is None:
             params = {"token": "unknown"}
 
-        if not os.environ.get("T_ANALYSIS_TOKEN") and params.get("token"):
-            os.environ["T_ANALYSIS_TOKEN"] = params["token"]
-
-        # Configure runtime region
-        runtimeRegion = params.get("region") if getRegionObj(params.get("region")) else None
-        if runtimeRegion:
-            setRuntimeRegion(runtimeRegion)
-
-        return Analysis(analysis, params)
+        return Analysis(params).init(analysis)

src/tagoio_sdk/modules/Analysis/Analysis_Type.py

@@ -45,7 +45,7 @@ class AnalysisConstructorParams(TypedDict, total=False):
 AnalysisID = str
 
 
-class TagoContext(TypedDict):
+class TagoContext:
     """
     TagoIO Analysis Context interface.
     As current version of the SDK doesn't provide the full TagoContext interface.

src/tagoio_sdk/modules/Resources/Analyses.py

@@ -343,10 +343,10 @@ def listSnippets(self, runtime: SnippetRuntime) -> SnippetsListResponse:
         @example:
             ```python
             resources = Resources()
-            deno_snippets = resources.analyses.listSnippets("deno-rt2025")
+            python_snippets = resources.analyses.listSnippets("python-rt2025")
 
             # Print all snippet titles
-            for snippet in deno_snippets["snippets"]:
+            for snippet in python_snippets["snippets"]:
                 print(f"{snippet['title']}: {snippet['description']}")
             ```