ju dev notes

[thorwhalen]

Dev notes on ju stuff.

[thorwhalen]

pydantic schema-to-code (via system call, and with ingress and egress)

class SchemaModelGenerator:
    """
    A wrapper around datamodel-code-generator with ingress/egress transformations
    to handle problematic schemas.
    """
    
    def __init__(self, 
                 ingress_transform: Optional[Callable] = None,
                 egress_transform: Optional[Callable] = None):
        """
        Initialize with optional transformation functions.
        
        Args:
            ingress_transform: Function to transform schema before generation
            egress_transform: Function to transform generated code
        """
        self.ingress_transform = ingress_transform or (lambda x: x)
        self.egress_transform = egress_transform or (lambda x: x)
    
    def generate_models(self, 
                       schema: Dict[str, Any], 
                       output_file: str = "models.py") -> str:
        """
        Generate Pydantic models from JSON schema.
        
        Args:
            schema: JSON schema dictionary
            output_file: Output file path for generated models
            
        Returns:
            Generated code as string
        """
        # Apply ingress transformation
        transformed_schema = self.ingress_transform(schema)
        
        # Write schema to temp file (more reliable than stdin)
        with tempfile.NamedTemporaryFile(mode='w', suffix='.json', delete=False) as f:
            json.dump(transformed_schema, f, indent=2)
            schema_file = f.name
        
        try:
            # Use subprocess instead of the Python API to avoid hanging
            cmd = [
                "datamodel-codegen",
                "--input", schema_file,
                "--output", output_file,
                "--input-file-type", "json",
                "--target-python-version", "3.10",
                "--use-default",
                "--use-schema-description",
                "--field-constraints",
            ]
            
            result = subprocess.run(
                cmd,
                capture_output=True,
                text=True,
                timeout=10  # 10 second timeout to prevent hanging
            )
            
            if result.returncode != 0:
                raise RuntimeError(f"Generation failed: {result.stderr}")
            
            # Read the generated code
            with open(output_file, 'r') as f:
                generated_code = f.read()
            
            # Apply egress transformation
            final_code = self.egress_transform(generated_code)
            
            # Write back the transformed code
            with open(output_file, 'w') as f:
                f.write(final_code)
            
            return final_code
            
        finally:
            # Clean up temp file
            Path(schema_file).unlink(missing_ok=True)