diff --git a/parallel_web_tools/__init__.py b/parallel_web_tools/__init__.py index eb41914..ea07545 100644 --- a/parallel_web_tools/__init__.py +++ b/parallel_web_tools/__init__.py @@ -27,7 +27,7 @@ run_tasks, ) -__version__ = "0.0.3" +__version__ = "0.0.4" __all__ = [ # Auth diff --git a/parallel_web_tools/integrations/bigquery/cloud_function/main.py b/parallel_web_tools/integrations/bigquery/cloud_function/main.py index e5d185c..1ae2e09 100644 --- a/parallel_web_tools/integrations/bigquery/cloud_function/main.py +++ b/parallel_web_tools/integrations/bigquery/cloud_function/main.py @@ -175,22 +175,22 @@ def parallel_enrich(request: Request): logger.info(f"Using batch processing for {len(inputs)} inputs") batch_results = _process_batch(inputs, common_output_cols, processor) - replies = [""] * len(calls) + replies: list[dict[str, Any] | None] = [None] * len(calls) for error_call in error_calls: - replies[error_call["index"]] = json.dumps({"error": error_call["error"]}) + replies[error_call["index"]] = {"error": error_call["error"]} for j, valid_call in enumerate(valid_calls): - replies[valid_call["index"]] = json.dumps(batch_results[j]) + replies[valid_call["index"]] = batch_results[j] return jsonify({"replies": replies}) # Fallback: process individually - replies = [""] * len(calls) + replies: list[dict[str, Any] | None] = [None] * len(calls) for error_call in error_calls: - replies[error_call["index"]] = json.dumps({"error": error_call["error"]}) + replies[error_call["index"]] = {"error": error_call["error"]} for valid_call in valid_calls: result = _process_batch([valid_call["input_data"]], valid_call["output_columns"], processor) - replies[valid_call["index"]] = json.dumps(result[0] if result else {"error": "No result"}) + replies[valid_call["index"]] = result[0] if result else {"error": "No result"} return jsonify({"replies": replies}) diff --git a/parallel_web_tools/integrations/bigquery/cloud_function/requirements.txt b/parallel_web_tools/integrations/bigquery/cloud_function/requirements.txt index c82b529..fe3b239 100644 --- a/parallel_web_tools/integrations/bigquery/cloud_function/requirements.txt +++ b/parallel_web_tools/integrations/bigquery/cloud_function/requirements.txt @@ -1,5 +1,5 @@ # Cloud Function dependencies for BigQuery Remote Function functions-framework>=3.0.0 flask>=3.0.0 -parallel-web-tools>=0.0.3 +parallel-web-tools>=0.0.4 google-cloud-secret-manager>=2.20.0 diff --git a/parallel_web_tools/integrations/bigquery/deploy.py b/parallel_web_tools/integrations/bigquery/deploy.py index 4704f85..099ed71 100644 --- a/parallel_web_tools/integrations/bigquery/deploy.py +++ b/parallel_web_tools/integrations/bigquery/deploy.py @@ -19,7 +19,9 @@ def _run_command(cmd: list[str], check: bool = True) -> subprocess.CompletedProc """Run a shell command and return the result.""" result = subprocess.run(cmd, capture_output=True, text=True) if check and result.returncode != 0: - raise RuntimeError(f"Command failed: {' '.join(cmd)}\n{result.stderr}") + # Include both stdout and stderr as some tools output errors to stdout + error_output = result.stderr or result.stdout or "(no output)" + raise RuntimeError(f"Command failed: {' '.join(cmd)}\n{error_output}") return result @@ -112,6 +114,21 @@ def deploy_bigquery_integration( Raises: RuntimeError: If any deployment step fails or user declines confirmation. """ + # Check for required CLI tools + import shutil + + missing_tools = [] + if not shutil.which("gcloud"): + missing_tools.append("gcloud") + if not shutil.which("bq"): + missing_tools.append("bq") + + if missing_tools: + raise RuntimeError( + f"Required CLI tools not found: {', '.join(missing_tools)}\n" + "Please install the Google Cloud SDK: https://cloud.google.com/sdk/docs/install" + ) + print(f"Checking for existing resources in {project_id}...") # Check for existing resources @@ -150,10 +167,17 @@ def deploy_bigquery_integration( result = _run_command(["gcloud", "secrets", "describe", secret_name, "--project", project_id], check=False) if result.returncode == 0: - # Update existing secret - _run_command( - ["gcloud", "secrets", "versions", "add", secret_name, "--data-file=-", "--project", project_id], check=True + # Update existing secret - use Popen to provide api_key via stdin + process = subprocess.Popen( + ["gcloud", "secrets", "versions", "add", secret_name, "--data-file=-", "--project", project_id], + stdin=subprocess.PIPE, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + text=True, ) + stdout, stderr = process.communicate(input=api_key) + if process.returncode != 0: + raise RuntimeError(f"Failed to update secret: {stderr}") else: # Create new secret process = subprocess.Popen( @@ -172,7 +196,9 @@ def deploy_bigquery_integration( stderr=subprocess.PIPE, text=True, ) - process.communicate(input=api_key) + stdout, stderr = process.communicate(input=api_key) + if process.returncode != 0: + raise RuntimeError(f"Failed to create secret: {stderr}") secret_resource = f"projects/{project_id}/secrets/{secret_name}/versions/latest" @@ -187,7 +213,7 @@ def deploy_bigquery_integration( "deploy", function_name, "--gen2", - "--runtime=python311", + "--runtime=python312", "--region", region, "--source", @@ -274,8 +300,8 @@ def deploy_bigquery_integration( "bq", "show", "--connection", - f"{project_id}.{region}.{connection_id}", "--format=json", + f"{project_id}.{region}.{connection_id}", ] ) connection_info = json.loads(result.stdout) @@ -333,10 +359,24 @@ def deploy_bigquery_integration( ) # Execute each CREATE FUNCTION statement - for statement in sql.split(";"): - statement = statement.strip() - if statement.startswith("CREATE"): - _run_command(["bq", "query", "--use_legacy_sql=false", statement + ";"]) + # Strip SQL comments (lines starting with --) before checking for CREATE + def strip_sql_comments(sql_text: str) -> str: + lines = [line for line in sql_text.split("\n") if not line.strip().startswith("--")] + return "\n".join(lines).strip() + + statements = [] + for chunk in sql.split(";"): + clean = strip_sql_comments(chunk) + if clean.startswith("CREATE"): + statements.append(clean) + + if not statements: + print(" Warning: No CREATE statements found in SQL template") + for statement in statements: + # Extract function name for logging + func_name = statement.split("`")[1] if "`" in statement else "unknown" + print(f" Creating {func_name}...") + _run_command(["bq", "query", "--use_legacy_sql=false", f"--project_id={project_id}", statement + ";"]) print("\nDeployment complete!") @@ -346,10 +386,24 @@ def deploy_bigquery_integration( "dataset_id": dataset_id, "function_url": function_url, "example_query": f""" +-- Basic usage (returns JSON with enriched fields and basis/citations) SELECT `{project_id}.{dataset_id}.parallel_enrich`( JSON_OBJECT('company_name', 'Google', 'website', 'google.com'), JSON_ARRAY('CEO name', 'Founding year', 'Brief description') ) as enriched_data; + +-- Parsing the JSON result into columns +SELECT + JSON_VALUE(enriched_data, '$.ceo_name') as ceo_name, + JSON_VALUE(enriched_data, '$.founding_year') as founding_year, + JSON_VALUE(enriched_data, '$.brief_description') as description, + JSON_QUERY(enriched_data, '$.basis') as citations +FROM ( + SELECT `{project_id}.{dataset_id}.parallel_enrich`( + JSON_OBJECT('company_name', 'Google', 'website', 'google.com'), + JSON_ARRAY('CEO name', 'Founding year', 'Brief description') + ) as enriched_data +); """.strip(), } diff --git a/parallel_web_tools/integrations/bigquery/sql/create_functions.sql b/parallel_web_tools/integrations/bigquery/sql/create_functions.sql index c7195af..7d1b016 100644 --- a/parallel_web_tools/integrations/bigquery/sql/create_functions.sql +++ b/parallel_web_tools/integrations/bigquery/sql/create_functions.sql @@ -10,40 +10,14 @@ -- {function_url} - Deployed Cloud Function URL -- Main enrichment function +-- Accepts JSON input and returns JSON output CREATE OR REPLACE FUNCTION `{project_id}.{dataset_id}.parallel_enrich`( - input_data STRING, - output_columns STRING + input_data JSON, + output_columns JSON ) -RETURNS STRING +RETURNS JSON REMOTE WITH CONNECTION `{project_id}.{location}.{connection_id}` OPTIONS ( endpoint = '{function_url}', user_defined_context = [("processor", "lite-fast")] ); - --- Convenience function for company enrichment -CREATE OR REPLACE FUNCTION `{project_id}.{dataset_id}.parallel_enrich_company`( - company_name STRING, - company_website STRING, - fields STRING -) -RETURNS STRING -AS ( - `{project_id}.{dataset_id}.parallel_enrich`( - JSON_OBJECT('company_name', company_name, 'website', company_website), - fields - ) -); - --- Example queries: --- --- SELECT parallel_enrich( --- JSON_OBJECT('company_name', 'Google', 'website', 'google.com'), --- JSON_ARRAY('CEO name', 'Founding year', 'Brief description') --- ); --- --- SELECT parallel_enrich_company( --- 'Apple', --- 'apple.com', --- JSON_ARRAY('CEO name', 'Market cap', 'Industry') --- ); diff --git a/pyproject.toml b/pyproject.toml index f48848d..4441f8b 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "hatchling.build" [project] name = "parallel-web-tools" -version = "0.0.3" +version = "0.0.4" description = "Parallel Tools: CLI and data enrichment utilities for the Parallel API" readme = "README.md" requires-python = ">=3.12" @@ -88,6 +88,15 @@ dev = [ [tool.hatch.build.targets.wheel] packages = ["parallel_web_tools"] +artifacts = [ + "*.sql", + "*.txt", # requirements.txt for cloud function +] + +[tool.hatch.build.targets.sdist] +include = [ + "/parallel_web_tools", +] [tool.pyrefly] project-includes = [ diff --git a/tests/test_cli.py b/tests/test_cli.py index 7ceaf9d..d70098e 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -112,7 +112,7 @@ def test_version(self, runner): """Should show version.""" result = runner.invoke(main, ["--version"]) assert result.exit_code == 0 - assert "0.0.3" in result.output + assert "0.0.4" in result.output class TestAuthCommand: diff --git a/uv.lock b/uv.lock index f901fff..29ab4ef 100644 --- a/uv.lock +++ b/uv.lock @@ -1057,7 +1057,7 @@ wheels = [ [[package]] name = "parallel-web-tools" -version = "0.0.3" +version = "0.0.4" source = { editable = "." } dependencies = [ { name = "pandas" },