# Python Upload

This Python script uploads a zip file to Euno's endpoint.

## Prerequisites

* Python 3.6+
* `requests` library (`pip install requests`)

## Script

```python
import requests
import os
import json
from pathlib import Path


# Configuration - replace with your values
endpoint_url = "https://api.app.euno.ai/accounts/YOUR_ACCOUNT_ID/integrations/YOUR_INTEGRATION_ID/prepare-upload"
trigger_secret = "your_trigger_secret_here"  # Trigger secret from your dbt source settings
zip_file_path = "Archive.zip"  # Path to your zip file

# Headers with authorization token
headers = {
    "authorization": f"Bearer {trigger_secret}",
    "content-type": "application/json"
}


def get_signed_url():
    """Request a signed upload URL for the zip file."""
    body = {"filename": Path(zip_file_path).name}
    response = requests.post(endpoint_url, headers=headers, json=body, timeout=30)
    
    if response.status_code != 200:
        print(f"Failed to obtain signed URL (status {response.status_code}): {response.text}")
        return None
    
    payload = response.json()
    upload_url = (
        payload.get("body", {}).get("upload", {}).get("url")
    )
    
    if not upload_url:
        print("Signed upload URL not found in response:", json.dumps(payload)[:500])
        return None
    
    return upload_url, payload


def upload_file(upload_url):
    """Upload the zip file to the signed URL."""
    try:
        with open(zip_file_path, "rb") as fp:
            response = requests.put(upload_url, data=fp, headers={"content-type": "application/zip"}, timeout=60)
        
        if response.status_code not in (200, 201):
            print(f"Upload failed (status {response.status_code}): {response.text[:200]}")
            return False
        
        return True
        
    except Exception as e:
        print("An error occurred during upload:", e)
        return False


def main():
    """Main function to upload archive using prepare-upload flow."""
    print("Starting archive upload process...")
    
    # Check if file exists
    if not os.path.exists(zip_file_path):
        print(f"Error: file '{zip_file_path}' not found.")
        return False
    
    # Get signed URL
    print("Requesting signed URL...")
    result = get_signed_url()
    if not result:
        return False
    
    upload_url, meta = result
    print("Signed URL obtained — uploading archive...")
    
    # Upload file
    success = upload_file(upload_url)
    
    if success:
        print("Upload succeeded! Prepare-upload response:")
        print(json.dumps(meta, indent=2))
        return True
    else:
        print("Upload failed.")
        return False


if __name__ == "__main__":
    main()
```

## Usage

1. **Configure the script**: Update the `endpoint_url`, `trigger_secret`, and `zip_file_path` variables with your specific values
2. **Install dependencies**: `pip install requests`
3. **Run the script**: Execute from your dbt project's `target` directory where the artifact files are located
4. **Verify upload**: Check the script output for success/failure messages


---

# Agent Instructions: Querying This Documentation

If you need additional information that is not directly available in this page, you can query the documentation dynamically by asking a question.

Perform an HTTP GET request on the current page URL with the `ask` query parameter:

```
GET https://docs.euno.ai/sources/transformation-etl/dbt-core/python-upload.md?ask=<question>
```

The question should be specific, self-contained, and written in natural language.
The response will contain a direct answer to the question and relevant excerpts and sources from the documentation.

Use this mechanism when the answer is not explicitly present in the current page, you need clarification or additional context, or you want to retrieve related documentation sections.