topics.py

"""Topic management endpoints"""
from fastapi import APIRouter, HTTPException, status, Depends
from confluent_kafka.admin import NewTopic, ConfigResource, ResourceType, KafkaException
from auth import get_current_user, User
from models import TopicCreate, TopicConfigUpdate
from services.kafka_admin import get_admin_client
import logging

logger = logging.getLogger(__name__)
router = APIRouter(prefix="/topics", tags=["Topics"])

@router.post("", status_code=status.HTTP_201_CREATED)
async def create_topic(topic: TopicCreate, current_user: User = Depends(get_current_user)):
    """Create a new topic in the Kafka cluster"""
    try:
        admin_client = get_admin_client()
        
        # Check if topic already exists
        metadata = admin_client.list_topics(timeout=10)
        if topic.name in metadata.topics:
            raise HTTPException(
                status_code=status.HTTP_409_CONFLICT,
                detail=f"Topic '{topic.name}' already exists"
            )
        
        # Create new topic
        new_topic = NewTopic(
            topic=topic.name,
            num_partitions=topic.num_partitions,
            replication_factor=topic.replication_factor,
            config=topic.config or {}
        )
        
        # Execute topic creation
        futures = admin_client.create_topics([new_topic])
        
        # Wait for the operation to complete
        for topic_name, future in futures.items():
            try:
                future.result(timeout=10)
                logger.info(f"Topic '{topic_name}' created successfully")
            except KafkaException as e:
                logger.error(f"Failed to create topic '{topic_name}': {str(e)}")
                raise HTTPException(
                    status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
                    detail=f"Failed to create topic: {str(e)}"
                )
        
        return {
            "message": f"Topic '{topic.name}' created successfully",
            "topic": {
                "name": topic.name,
                "num_partitions": topic.num_partitions,
                "replication_factor": topic.replication_factor,
                "config": topic.config or {}
            }
        }
    
    except HTTPException:
        raise
    except Exception as e:
        logger.error(f"Failed to create topic: {str(e)}")
        raise HTTPException(
            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
            detail=f"Failed to create topic: {str(e)}"
        )

@router.get("")
async def list_topics(current_user: User = Depends(get_current_user)):
    """List all topics in the Kafka cluster"""
    try:
        admin_client = get_admin_client()
        metadata = admin_client.list_topics(timeout=10)
        
        topics = []
        for topic_name, topic_metadata in metadata.topics.items():
            if not topic_name.startswith('_'):
                partitions = topic_metadata.partitions
                replication_factor = len(partitions[0].replicas) if partitions else 0
                
                topics.append({
                    "name": topic_name,
                    "partition_count": len(partitions),
                    "replication_factor": replication_factor
                })
        
        return {
            "count": len(topics),
            "topics": topics
        }
    
    except Exception as e:
        logger.error(f"Failed to list topics: {str(e)}")
        raise HTTPException(
            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
            detail=f"Failed to list topics: {str(e)}"
        )

@router.get("/{topic_name}")
async def describe_topic(topic_name: str, current_user: User = Depends(get_current_user)):
    """Get detailed information about a specific topic"""
    try:
        admin_client = get_admin_client()
        
        metadata = admin_client.list_topics(topic=topic_name, timeout=10)
        
        if topic_name not in metadata.topics:
            raise HTTPException(
                status_code=status.HTTP_404_NOT_FOUND,
                detail=f"Topic '{topic_name}' not found"
            )
        
        topic_metadata = metadata.topics[topic_name]
        
        partitions = []
        for partition_id, partition_metadata in topic_metadata.partitions.items():
            partitions.append({
                "partition_id": partition_id,
                "leader": partition_metadata.leader,
                "replicas": partition_metadata.replicas,
                "isr": partition_metadata.isrs
            })
        
        config_resource = ConfigResource(ResourceType.TOPIC, topic_name)
        configs_result = admin_client.describe_configs([config_resource])
        
        topic_configs = {}
        for res, future in configs_result.items():
            try:
                config_entries = future.result(timeout=10)
                for config_name, config_entry in config_entries.items():
                    topic_configs[config_name] = {
                        "value": config_entry.value,
                        "source": str(config_entry.source),
                        "is_default": config_entry.is_default,
                        "is_sensitive": config_entry.is_sensitive
                    }
            except Exception as e:
                logger.warning(f"Failed to get configs for topic '{topic_name}': {str(e)}")
        
        replication_factor = len(partitions[0]["replicas"]) if partitions else 0
        
        return {
            "name": topic_name,
            "partition_count": len(partitions),
            "replication_factor": replication_factor,
            "partitions": partitions,
            "config": topic_configs
        }
    
    except HTTPException:
        raise
    except Exception as e:
        logger.error(f"Failed to describe topic '{topic_name}': {str(e)}")
        raise HTTPException(
            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
            detail=f"Failed to describe topic: {str(e)}"
        )

@router.put("/{topic_name}")
async def alter_topic_config(topic_name: str, config_update: TopicConfigUpdate, current_user: User = Depends(get_current_user)):
    """Alter topic-level configuration properties"""
    try:
        admin_client = get_admin_client()
        
        metadata = admin_client.list_topics(topic=topic_name, timeout=10)
        if topic_name not in metadata.topics:
            raise HTTPException(
                status_code=status.HTTP_404_NOT_FOUND,
                detail=f"Topic '{topic_name}' not found"
            )
        
        config_resource = ConfigResource(
            ResourceType.TOPIC,
            topic_name,
            set_config=config_update.config
        )
        
        futures = admin_client.alter_configs([config_resource])
        
        for res, future in futures.items():
            try:
                future.result(timeout=10)
                logger.info(f"Topic '{topic_name}' configuration updated successfully")
            except KafkaException as e:
                logger.error(f"Failed to update topic '{topic_name}' configuration: {str(e)}")
                raise HTTPException(
                    status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
                    detail=f"Failed to update topic configuration: {str(e)}"
                )
        
        return {
            "message": f"Topic '{topic_name}' configuration updated successfully",
            "topic": topic_name,
            "updated_config": config_update.config
        }
    
    except HTTPException:
        raise
    except Exception as e:
        logger.error(f"Failed to alter topic configuration: {str(e)}")
        raise HTTPException(
            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
            detail=f"Failed to alter topic configuration: {str(e)}"
        )