Python Integration

Complete Python examples for Lunar Stream API integration.

Installation

pip install requests python-dotenv

API Client

# lunar_stream_client.py
import os
import requests
from typing import Optional, List, Dict, Any

class LunarStreamClient:
    def __init__(self, api_key: str, project_id: str):
        self.project_id = project_id
        self.base_url = os.getenv('LUNAR_STREAM_API_URL', 'https://api.lunarstream.kozow.com/api/v1')
        self.headers = {
            'ls-api-key': api_key,
            'Content-Type': 'application/json'
        }

    def get_media_servers(self) -> List[Dict[str, Any]]:
        response = requests.get(
            f'{self.base_url}/media-servers/available',
            headers=self.headers,
            params={'projectId': self.project_id}
        )
        response.raise_for_status()
        return response.json()['data']

    def create_livestream(self, name: str, media_server_id: str) -> Dict[str, Any]:
        response = requests.post(
            f'{self.base_url}/livestream',
            headers=self.headers,
            json={
                'name': name,
                'projectId': self.project_id,
                'mediaServerId': media_server_id
            }
        )
        response.raise_for_status()
        return response.json()['data']

    def get_livestream(self, livestream_id: str) -> Dict[str, Any]:
        response = requests.get(
            f'{self.base_url}/livestream/{livestream_id}',
            headers=self.headers,
            params={'projectId': self.project_id}
        )
        response.raise_for_status()
        return response.json()['data']

    def list_livestreams(self, page: int = 1, limit: int = 10) -> Dict[str, Any]:
        response = requests.get(
            f'{self.base_url}/livestream',
            headers=self.headers,
            params={'projectId': self.project_id, 'page': page, 'limit': limit}
        )
        response.raise_for_status()
        return response.json()

    def delete_livestream(self, livestream_id: str) -> Dict[str, Any]:
        response = requests.delete(
            f'{self.base_url}/livestream/{livestream_id}',
            headers=self.headers,
            params={'projectId': self.project_id}
        )
        response.raise_for_status()
        return response.json()

    def generate_push_token(self, stream_key: str, expires_in: int = 3600) -> Dict[str, Any]:
        response = requests.post(
            f'{self.base_url}/livestream/generate-push-token',
            headers=self.headers,
            json={
                'projectId': self.project_id,
                'streamKey': stream_key,
                'expiresIn': expires_in
            }
        )
        response.raise_for_status()
        return response.json()['data']

FastAPI Example

# main.py
import os
from dotenv import load_dotenv
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from lunar_stream_client import LunarStreamClient

load_dotenv()

app = FastAPI()

client = LunarStreamClient(
    api_key=os.getenv('LUNAR_STREAM_API_KEY'),
    project_id=os.getenv('LUNAR_STREAM_PROJECT_ID')
)

class CreateStreamRequest(BaseModel):
    name: str

@app.post('/api/streams')
async def create_stream(request: CreateStreamRequest):
    try:
        servers = client.get_media_servers()
        if not servers:
            raise HTTPException(status_code=503, detail='No media servers available')

        stream = client.create_livestream(request.name, servers[0]['id'])

        return {
            'id': stream['id'],
            'name': stream['name'],
            'rtmpUrl': f"{stream['rmtpServer']}/{stream['streamKeyWithParams']}",
            'hlsUrl': stream['hlsSources'][0]['url'] if stream.get('hlsSources') else None,
            'status': stream['status']
        }
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

@app.get('/api/streams/{stream_key}')
async def get_stream(stream_key: str):
    try:
        import requests
        response = requests.get(
            f'https://api.lunarstream.kozow.com/api/v1/livestreams/stream-info/{stream_key}'
        )
        response.raise_for_status()
        stream = response.json()['data']

        return {
            'name': stream['name'],
            'status': stream['status'],
            'hlsUrls': stream.get('hlsSources', [])
        }
    except:
        raise HTTPException(status_code=404, detail='Stream not found')

Usage

# Run server
uvicorn main:app --reload

# Create stream
curl -X POST http://localhost:8000/api/streams \
  -H "Content-Type: application/json" \
  -d '{"name": "My Stream"}'