Quick Start - Python

Get started with Importly in your Python application using popular libraries and frameworks.

Setup

Install the required dependencies:

bash
Copy
1pip install requests python-dotenv

For async support (optional):

bash
Copy
1pip install aiohttp asyncio

1. Get Your API Key

1. Create an Account at Importly.io
2. Find your API Key via the api key page
3. Set it as an environment variable:

bash
Copy
1# .env
2IMPORTLY_API_KEY=your_api_key_here
3IMPORTLY_API_URL=https://api.importly.io
4WEBHOOK_URL=https://your-domain.com/webhook/importly

2. Create Importly Client

Create a synchronous client using requests:

python
Copy
1# importly_client.py
2import os
3import time
4import requests
5from typing import Optional, Dict, Any
6from dotenv import load_dotenv
7
8load_dotenv()
9
10class ImportlyClient:
11    def __init__(self, api_key: Optional[str] = None):
12        self.api_key = api_key or os.getenv('IMPORTLY_API_KEY')
13        if not self.api_key:
14            raise ValueError("API key is required")
15
16        self.base_url = os.getenv('IMPORTLY_API_URL')
17        self.session = requests.Session()
18        self.session.headers.update({
19            'Authorization': f'Bearer {self.api_key}',
20            'Content-Type': 'application/json'
21        })
22        self.session.timeout = 30
23
24    def import_media(self, url: str, **options) -> Dict[str, Any]:
25        """Import media from a URL"""
26        data = {
27            'url': url,
28            'includeVideo': options.get('include_video', True),
29            'includeAudio': options.get('include_audio', True),
30            'videoQuality': options.get('video_quality', '1080p'),
31            'audioQuality': options.get('audio_quality', 'medium'),
32        }
33
34        webhook_url = options.get('webhook_url')
35        if webhook_url:
36            data['webhookUrl'] = webhook_url
37
38        try:
39            response = self.session.post(f"{self.base_url}/import", json=data)
40            response.raise_for_status()
41            return response.json()
42        except requests.RequestException as e:
43            raise self._handle_error(e)
44
45    def get_metadata(self, url: str, webhook_url: Optional[str] = None) -> Dict[str, Any]:
46        """Get metadata for a URL"""
47        params = {'url': url}
48        if webhook_url:
49            params['webhookUrl'] = webhook_url
50
51        try:
52            response = self.session.get(f"{self.base_url}/metadata", params=params)
53            response.raise_for_status()
54            return response.json()
55        except requests.RequestException as e:
56            raise self._handle_error(e)
57
58    def check_import_status(self, import_id: str) -> Dict[str, Any]:
59        """Check the status of an import"""
60        try:
61            response = self.session.get(f"{self.base_url}/import/status",
62                                      params={'id': import_id})
63            response.raise_for_status()
64            return response.json()
65        except requests.RequestException as e:
66            raise self._handle_error(e)
67
68    def check_metadata_status(self, job_id: str) -> Dict[str, Any]:
69        """Check the status of a metadata job"""
70        try:
71            response = self.session.get(f"{self.base_url}/metadata/status",
72                                      params={'id': job_id})
73            response.raise_for_status()
74            return response.json()
75        except requests.RequestException as e:
76            raise self._handle_error(e)
77
78    def wait_for_completion(self, id: str, job_type: str = 'import',
79                          max_wait_time: int = 300, poll_interval: int = 5) -> Dict[str, Any]:
80        """Wait for an import or metadata job to complete"""
81        start_time = time.time()
82        check_status = self.check_import_status if job_type == 'import' else self.check_metadata_status
83
84        while time.time() - start_time < max_wait_time:
85            try:
86                result = check_status(id)
87                status = result['data']['status']
88
89                if status == 'completed':
90                    return result
91                elif status in ['failed', 'cancelled']:
92                    error_msg = result['data'].get('error', 'Unknown error')
93                    raise Exception(f"{job_type} {status}: {error_msg}")
94
95                time.sleep(poll_interval)
96            except requests.RequestException as e:
97                if hasattr(e, 'response') and e.response.status_code == 404:
98                    raise Exception(f"{job_type} not found")
99                raise
100
101        raise Exception(f"{job_type} timed out after {max_wait_time} seconds")
102
103    def _handle_error(self, error: requests.RequestException) -> Exception:
104        """Handle and format API errors"""
105        if hasattr(error, 'response') and error.response is not None:
106            status_code = error.response.status_code
107            try:
108                error_data = error.response.json()
109                message = error_data.get('message', error_data.get('error', 'API request failed'))
110            except ValueError:
111                message = error.response.text or 'API request failed'
112
113            if status_code == 401:
114                return Exception('Invalid API key')
115            elif status_code == 402:
116                return Exception('Insufficient credits')
117            elif status_code == 429:
118                return Exception('Rate limit exceeded')
119            elif status_code == 400:
120                return Exception(f'Bad request: {message}')
121            else:
122                return Exception(f'API error ({status_code}): {message}')
123        else:
124            return Exception(f'Network error: {str(error)}')

3. Async Client (Optional)

For async applications, create an async client:

python
Copy
1# async_importly_client.py
2import os
3import asyncio
4import aiohttp
5from typing import Optional, Dict, Any
6from dotenv import load_dotenv
7
8load_dotenv()
9
10class AsyncImportlyClient:
11    def __init__(self, api_key: Optional[str] = None):
12        self.api_key = api_key or os.getenv('IMPORTLY_API_KEY')
13        if not self.api_key:
14            raise ValueError("API key is required")
15
16        self.base_url = "https://api.importly.io"
17        self.headers = {
18            'Authorization': f'Bearer {self.api_key}',
19            'Content-Type': 'application/json'
20        }
21
22    async def _make_request(self, method: str, endpoint: str, **kwargs) -> Dict[str, Any]:
23        """Make an HTTP request"""
24        url = f"{self.base_url}{endpoint}"
25
26        async with aiohttp.ClientSession(headers=self.headers, timeout=aiohttp.ClientTimeout(30)) as session:
27            async with session.request(method, url, **kwargs) as response:
28                if response.status >= 400:
29                    error_text = await response.text()
30                    raise aiohttp.ClientError(f"HTTP {response.status}: {error_text}")
31
32                return await response.json()
33
34    async def import_media(self, url: str, **options) -> Dict[str, Any]:
35        """Import media from a URL"""
36        data = {
37            'url': url,
38            'includeVideo': options.get('include_video', True),
39            'includeAudio': options.get('include_audio', True),
40            'videoQuality': options.get('video_quality', '1080p'),
41            'audioQuality': options.get('audio_quality', 'medium'),
42        }
43
44        webhook_url = options.get('webhook_url')
45        if webhook_url:
46            data['webhookUrl'] = webhook_url
47
48        return await self._make_request('POST', '/import', json=data)
49
50    async def get_metadata(self, url: str, webhook_url: Optional[str] = None) -> Dict[str, Any]:
51        """Get metadata for a URL"""
52        params = {'url': url}
53        if webhook_url:
54            params['webhookUrl'] = webhook_url
55
56        return await self._make_request('GET', '/metadata', params=params)
57
58    async def check_import_status(self, import_id: str) -> Dict[str, Any]:
59        """Check the status of an import"""
60        return await self._make_request('GET', '/import/status', params={'id': import_id})
61
62    async def check_metadata_status(self, job_id: str) -> Dict[str, Any]:
63        """Check the status of a metadata job"""
64        return await self._make_request('GET', '/metadata/status', params={'id': job_id})
65
66    async def wait_for_completion(self, id: str, job_type: str = 'import',
67                                max_wait_time: int = 300, poll_interval: int = 5) -> Dict[str, Any]:
68        """Wait for an import or metadata job to complete"""
69        start_time = asyncio.get_event_loop().time()
70        check_status = self.check_import_status if job_type == 'import' else self.check_metadata_status
71
72        while asyncio.get_event_loop().time() - start_time < max_wait_time:
73            try:
74                result = await check_status(id)
75                status = result['data']['status']
76
77                if status == 'completed':
78                    return result
79                elif status in ['failed', 'cancelled']:
80                    error_msg = result['data'].get('error', 'Unknown error')
81                    raise Exception(f"{job_type} {status}: {error_msg}")
82
83                await asyncio.sleep(poll_interval)
84            except aiohttp.ClientError as e:
85                if '404' in str(e):
86                    raise Exception(f"{job_type} not found")
87                raise
88
89        raise Exception(f"{job_type} timed out after {max_wait_time} seconds")

4. Flask Web Application

Create a Flask app with webhook support:

python
Copy
1# app.py
2from flask import Flask, request, jsonify
3from importly_client import ImportlyClient
4import os
5from datetime import datetime
6
7app = Flask(__name__)
8
9# Initialize Importly client
10importly = ImportlyClient()
11
12# In-memory storage (use database in production)
13imports_db = {}
14metadata_db = {}
15
16@app.route('/import', methods=['POST'])
17def import_media():
18    """Start a media import"""
19    try:
20        data = request.get_json()
21        url = data.get('url')
22
23        if not url:
24            return jsonify({'error': 'URL is required'}), 400
25
26        # Start import with webhook
27        result = importly.import_media(
28            url,
29            video_quality=data.get('videoQuality', '1080p'),
30            audio_quality=data.get('audioQuality', 'medium'),
31            webhook_url=os.getenv('WEBHOOK_URL')
32        )
33
34        import_id = result['data']['jobId']
35
36        # Store import info
37        imports_db[import_id] = {
38            'id': import_id,
39            'url': url,
40            'status': 'queued',
41            'created_at': datetime.utcnow().isoformat(),
42            'video_quality': data.get('videoQuality', '1080p'),
43            'audio_quality': data.get('audioQuality', 'medium')
44        }
45
46        return jsonify({
47            'success': True,
48            'import_id': import_id,
49            'status': 'queued',
50            'message': 'Import started successfully'
51        })
52
53    except Exception as e:
54        return jsonify({'error': str(e)}), 500
55
56@app.route('/metadata', methods=['POST'])
57def get_metadata():
58    """Get metadata for a URL"""
59    try:
60        data = request.get_json()
61        url = data.get('url')
62
63        if not url:
64            return jsonify({'error': 'URL is required'}), 400
65
66        result = importly.get_metadata(url, os.getenv('WEBHOOK_URL'))
67        job_id = result['data']['jobId']
68
69        # Store metadata info
70        metadata_db[job_id] = {
71            'id': job_id,
72            'url': url,
73            'status': 'queued',
74            'created_at': datetime.utcnow().isoformat()
75        }
76
77        return jsonify({
78            'success': True,
79            'job_id': job_id,
80            'status': 'queued',
81            'message': 'Metadata request started successfully'
82        })
83
84    except Exception as e:
85        return jsonify({'error': str(e)}), 500
86
87@app.route('/import/<import_id>/status', methods=['GET'])
88def get_import_status(import_id):
89    """Get import status"""
90    try:
91        # Check local storage first
92        local_import = imports_db.get(import_id)
93        if local_import and local_import['status'] == 'completed':
94            return jsonify({'success': True, 'data': local_import})
95
96        # Check with Importly API
97        result = importly.check_import_status(import_id)
98
99        # Update local storage
100        if import_id in imports_db:
101            imports_db[import_id].update(result['data'])
102
103        return jsonify(result)
104
105    except Exception as e:
106        return jsonify({'error': str(e)}), 500
107
108@app.route('/metadata/<job_id>/status', methods=['GET'])
109def get_metadata_status(job_id):
110    """Get metadata status"""
111    try:
112        # Check local storage first
113        local_metadata = metadata_db.get(job_id)
114        if local_metadata and local_metadata['status'] == 'completed':
115            return jsonify({'success': True, 'data': local_metadata})
116
117        # Check with Importly API
118        result = importly.check_metadata_status(job_id)
119
120        # Update local storage
121        if job_id in metadata_db:
122            metadata_db[job_id].update(result['data'])
123
124        return jsonify(result)
125
126    except Exception as e:
127        return jsonify({'error': str(e)}), 500
128
129@app.route('/webhook/importly', methods=['POST'])
130def handle_webhook():
131    """Handle Importly webhooks (sent only on successful completion)"""
132    try:
133        body = request.get_json()
134        webhook_data = body.get('data', {})
135        webhook_type = webhook_data.get('type')
136
137        print(f"Received webhook: {webhook_type}")
138
139        if webhook_type == 'import':
140            handle_import_completed(webhook_data)
141        elif webhook_type == 'metadata':
142            handle_metadata_completed(webhook_data)
143        else:
144            print(f"Unknown webhook type: {webhook_type}")
145
146        return jsonify({'received': True})
147
148    except Exception as e:
149        print(f"Webhook error: {e}")
150        return jsonify({'error': 'Webhook processing failed'}), 500
151
152def handle_import_completed(data):
153    """Handle completed import webhook (result fields are flattened into data)"""
154    import_id = data['jobId']
155
156    if import_id in imports_db:
157        imports_db[import_id].update({
158            'status': 'completed',
159            'result': data,
160            'completed_at': datetime.utcnow().isoformat()
161        })
162
163    print(f"Import completed: {import_id}")
164
165    # Add your custom logic here:
166    # - Send notifications
167    # - Update database
168    # - Process the media file
169
170def handle_metadata_completed(data):
171    """Handle completed metadata webhook (result fields are flattened into data)"""
172    job_id = data['jobId']
173
174    if job_id in metadata_db:
175        metadata_db[job_id].update({
176            'status': 'completed',
177            'result': data,
178            'completed_at': datetime.utcnow().isoformat()
179        })
180
181    print(f"Metadata completed: {job_id}")
182
183@app.route('/imports', methods=['GET'])
184def list_imports():
185    """List all imports"""
186    imports_list = sorted(
187        imports_db.values(),
188        key=lambda x: x['created_at'],
189        reverse=True
190    )
191    return jsonify({'imports': imports_list})
192
193@app.route('/metadata', methods=['GET'])
194def list_metadata():
195    """List all metadata requests"""
196    metadata_list = sorted(
197        metadata_db.values(),
198        key=lambda x: x['created_at'],
199        reverse=True
200    )
201    return jsonify({'metadata': metadata_list})
202
203@app.route('/health', methods=['GET'])
204def health_check():
205    """Health check endpoint"""
206    return jsonify({
207        'status': 'ok',
208        'timestamp': datetime.utcnow().isoformat(),
209        'imports': len(imports_db),
210        'metadata': len(metadata_db)
211    })
212
213if __name__ == '__main__':
214    app.run(debug=True, port=5000)

5. Command Line Tool

Create a CLI tool for easy imports:

python
Copy
1# cli.py
2#!/usr/bin/env python3
3
4import sys
5import argparse
6from importly_client import ImportlyClient
7
8def main():
9    parser = argparse.ArgumentParser(description='Importly CLI tool')
10    parser.add_argument('url', help='URL to import')
11    parser.add_argument('--quality', '-q', default='1080p',
12                       choices=['480p', '720p', '1080p', '1440p', '2160p'],
13                       help='Video quality (default: 1080p)')
14    parser.add_argument('--audio-quality', '-a', default='medium',
15                       choices=['low', 'medium', 'high'],
16                       help='Audio quality (default: medium)')
17    parser.add_argument('--metadata-only', '-m', action='store_true',
18                       help='Only get metadata, don\'t import')
19    parser.add_argument('--no-wait', action='store_true',
20                       help='Don\'t wait for completion')
21
22    args = parser.parse_args()
23
24    try:
25        client = ImportlyClient()
26
27        if args.metadata_only:
28            print(f"Getting metadata for: {args.url}")
29            result = client.get_metadata(args.url)
30            job_id = result['data']['jobId']
31            print(f"Metadata job started with ID: {job_id}")
32
33            if not args.no_wait:
34                print("Waiting for completion...")
35                completed = client.wait_for_completion(job_id, 'metadata')
36                print("Metadata completed!")
37                print(f"Title: {completed['data']['result'].get('title', 'N/A')}")
38                print(f"Duration: {completed['data']['result'].get('duration', 'N/A')}s")
39        else:
40            print(f"Starting import for: {args.url}")
41            print(f"Video quality: {args.quality}")
42            print(f"Audio quality: {args.audio_quality}")
43
44            result = client.import_media(
45                args.url,
46                video_quality=args.quality,
47                audio_quality=args.audio_quality
48            )
49
50            import_id = result['data']['jobId']
51            print(f"Import started with ID: {import_id}")
52
53            if not args.no_wait:
54                print("Waiting for completion...")
55                completed = client.wait_for_completion(import_id, 'import')
56
57                print("Import completed!")
58                result_data = completed['data']['result']
59                print(f"Media URL: {result_data['mediaUrl']}")
60                print(f"Cost (USD): {result_data['costInDollars']}")
61                print(f"Duration: {result_data['duration']}s")
62                print(f"File size: {result_data['fileSizeBytes']} bytes")
63
64    except Exception as e:
65        print(f"Error: {e}", file=sys.stderr)
66        sys.exit(1)
67
68if __name__ == '__main__':
69    main()

Make it executable:

bash
Copy
1chmod +x cli.py

Use it:

bash
Copy
1python cli.py "https://example.com/video" --quality 1080p
2python cli.py "https://example.com/video" --metadata-only

6. Running the Application

Start your Flask app:

bash
Copy
1python app.py

Test the endpoints:

bash
Copy
1# Start an import
2curl -X POST http://localhost:5000/import \
3  -H "Content-Type: application/json" \
4  -d '{"url": "https://example.com/video", "videoQuality": "1080p"}'
5
6# Check status
7curl http://localhost:5000/import/YOUR_IMPORT_ID/status

Django Integration (Optional)

For Django projects, you can integrate Importly into your models and views:

python
Copy
1# models.py
2from django.db import models
3import uuid
4
5class MediaImport(models.Model):
6    STATUS_CHOICES = [
7        ('queued', 'Queued'),
8        ('processing', 'Processing'),
9        ('completed', 'Completed'),
10        ('failed', 'Failed'),
11    ]
12
13    id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False)
14    import_id = models.CharField(max_length=255, unique=True)
15    url = models.URLField()
16    status = models.CharField(max_length=20, choices=STATUS_CHOICES, default='queued')
17    video_quality = models.CharField(max_length=10, default='1080p')
18    audio_quality = models.CharField(max_length=10, default='medium')
19    media_url = models.URLField(blank=True, null=True)
20    credits_used = models.IntegerField(blank=True, null=True)
21    error_message = models.TextField(blank=True, null=True)
22    created_at = models.DateTimeField(auto_now_add=True)
23    completed_at = models.DateTimeField(blank=True, null=True)
24
25    def __str__(self):
26        return f"Import {self.import_id} - {self.status}"

Why Python?

Python is excellent for Importly because:

• ✅ Rich ecosystem - Great libraries for web development (Flask, Django, FastAPI)
• ✅ Data processing - Perfect for handling media metadata and files
• ✅ AI/ML integration - Easy to integrate with ML pipelines
• ✅ Async support - Handle multiple imports concurrently
• ✅ Simple syntax - Quick to implement and maintain

Best Practices

1. Use virtual environments to manage dependencies
2. Implement proper logging with Python's logging module
3. Use databases instead of in-memory storage for production
4. Add input validation with libraries like Pydantic
5. Implement caching with Redis for frequently accessed data
6. Use async/await for high-concurrency applications
7. Add comprehensive error handling and retries

Production Considerations

• Database: Use SQLAlchemy with PostgreSQL or similar
• Task Queue: Implement Celery with Redis/RabbitMQ for background jobs
• Web Framework: Consider FastAPI for async APIs or Django for full applications
• Monitoring: Add logging, metrics, and health checks
• Security: Implement authentication, input validation, and rate limiting

Complete Example

Check out our complete Python example on GitHub for a full implementation with Django/Flask integrations and advanced features.