Generate Python client methods from Retrofit definitions.

retropy.py

import argparse
import re
from pathlib import Path


def parse_keys(key: str, value: str) -> dict:
    """
    Parses Retrofit annotations (like @Path, @Query, etc.) and extracts their values as Python data types.

    Args:
        key (str): Annotation key (e.g., 'Path', 'Query', etc.).
        value (str): The string containing Retrofit annotations and method signature.

    Returns:
        dict: A dictionary mapping Retrofit annotation values to Python data types (str, int, bool).

    Raises:
        TypeError: If the type in the annotation is not recognized.
    """
    keys = {}
    obj = re.findall(rf'@{key}\("([^"]+)"\)\s+(\w+)\s+\w+', value)
    if obj:
        for k, v in obj:
            # Map Retrofit types to Python types
            if v == 'String':
                _type = 'str'
            elif v in ('int', 'Integer'):
                _type = 'int'
            elif v in ('bool', 'Boolean'):
                _type = 'bool'
            else:
                raise TypeError(v)

            keys[k] = _type
    return keys


def convert_to_camel_case(s: str) -> str:
    """
    Converts snake_case string to camelCase format, which is common in Python method naming conventions.

    Args:
        s (str): Input string in snake_case format.

    Returns:
        str: Output string in camelCase format.
    """
    return ''.join(word.capitalize() if i > 0 else word for i, word in enumerate(s.split('_')))


def parse_headers(line: str) -> dict:
    """
    Parses Retrofit header annotations (e.g., @Headers) from a line of Java code.

    Args:
        line (str): The line of Java code containing the Retrofit header annotations.

    Returns:
        dict: A dictionary containing parsed headers like Content-Type and others.
    """
    headers = {}
    if not line.startswith('@'):
        return headers

    # Parse content type annotations
    if '@FormUrlEncoded' in line:
        headers['Content-Type'] = 'application/x-www-form-urlencoded'
    elif '@Multipart' in line:
        headers['Content-Type'] = 'multipart/form-data'
    elif '@Headers(' in line:
        line = line.split('@Headers(')[1].rsplit(')')[0]
        if line.startswith('{"') and line.endswith('"}'):
            # Handle multiple headers in the format: {"Header1: value", "Header2: value"}
            items = line[1:-1].split(',')
            for item in items:
                key, value = item.strip()[1:-1].split(': ')
                headers[key] = value
        elif line.startswith('"') and line.endswith('"'):
            # Handle single header in the format: "Header: value"
            key, value = line[1:-1].split(': ')
            headers[key] = value

    return headers


class Retropy:
    """
    Class to parse Java Retrofit interface definitions and convert them into Python client methods.

    Attributes:
        domain (str): Base domain URL for the API requests.
        interfaces (list): List to store parsed interface data.
    """

    def __init__(self, domain: str):
        """
        Initializes Retropy with the provided domain and an empty list for interfaces.

        Args:
            domain (str): The base domain URL for API requests.
        """
        self.interfaces = []
        self.domain = domain

    def loads(self, data: str) -> None:
        """
        Loads and parses Retrofit interface data from a string.

        Args:
            data (str): The string content containing the Retrofit interface definitions.
        """
        self.interfaces.append(
            self.parse(data=data)
        )

    def load(self, path: Path) -> None:
        """
        Loads Retrofit interface data from a file or directory.

        Args:
            path (Path): Path to the file or directory containing Retrofit interface definitions.

        Raises:
            FileExistsError: If the file or directory is not found.
        """
        if path.is_file():
            self.loads(path.read_text())
        elif path.is_dir():
            for p in path.iterdir():
                self.loads(p.read_text())
        else:
            raise FileExistsError(path)

    def parse(self, data: str) -> dict:
        """
        Parses the Retrofit interface data and extracts method details such as headers, paths, query params, and more.

        Args:
            data (str): The string content containing Retrofit interface definitions.

        Returns:
            dict: A dictionary containing package, interface, and method details.
        """
        lines = data.splitlines()
        package, interface = None, None
        methods, headers = {}, {}

        for i, line in enumerate(lines):
            line = line.strip()

            # Parse package declaration
            obj = re.search(r'package\s+(.*);', line)
            if obj:
                package = obj.group(1)
                continue

            # Parse interface declaration
            obj = re.search(r'public interface\s+(\w+)\s+\{', line)
            if obj:
                interface = obj.group(1)
                continue

            if not line.startswith('@'):
                continue

            # Parse headers
            headers.update(parse_headers(line))

            # Parse method and URL
            obj = re.search(r'@(HTTP|GET|POST|PUT|PATCH|DELETE|OPTIONS|HEAD)\("([^"]+)"\)', line)
            if not obj:
                continue

            method, url = obj.groups()
            if url.startswith('/'):
                url = url[1:]

            # Find method definition
            outer = None
            for line in lines[i + 1:]:
                line = line.strip()
                # Parse headers
                headers.update(parse_headers(line))
                if not line.startswith('@'):
                    outer = line
                    break

            assert outer, 'Method definition not found'

            # Extract function name
            obj = re.search('\s+(\w+)\(', outer)
            assert obj, 'Function name not found'
            name = obj.group(1)

            # Generate method name if it's too short
            if len(name) < 3:
                name = convert_to_camel_case(f'{method.lower()}_{Path(url).name}')

            # Parse URL paths, query params, and body parts
            url_paths = parse_keys('Path', outer)
            url_params = parse_keys('Query', outer)
            body_field = parse_keys('Field', outer)
            body_part = parse_keys('Part', outer)
            body_headers = parse_keys('Header', outer)
            # body_data = get_keys('Body', outer)
            body_data = '@Body' in outer

            # Organize method details
            methods[name] = {
                'method': method,
                'url': url,
                'paths': {
                    **url_paths,
                    **url_params,
                    **body_field,
                    **body_part,
                    # **body_headers
                },
                'params': url_params,
                'data': {**body_field, **body_part} or (
                    {} if body_data or 'multipart' in headers.get('Content-Type', '') else None),
                'headers': {**headers, **body_headers}
            }
            headers = {}

        return {'package': package, 'interface': interface, 'methods': methods}

    def dump(self) -> str:
        """
        Converts parsed Retrofit interface data into Python client methods and outputs as a string.

        Returns:
            str: Generated Python client code as a string.
        """
        content = []

        for interface in self.interfaces:
            content.append('# package {package}.{interface};'.format(
                package=interface['package'],
                interface=interface['interface']
            ))

            for name, value in interface['methods'].items():
                args = ', '.join([f'{k}: {v}' for k, v in value['paths'].items()])
                content.append(f'def {name}(self' + (f", {args}" if args else '') + '):')

                headers = value['headers']
                data = value['data']
                is_multi = 'multipart' in headers.get('Content-Type', '')

                # Handle multipart form data
                if is_multi:
                    content.append('\tmp_encoder = MultipartEncoder(fields={')
                    content[-1] += ', '.join([f"'{k}': {k}" for k in data.keys()])
                    content[-1] += '})\n'

                # Build the request method call
                content.append('\treturn self.__request(')
                content.append(f"\t\tmethod='{value['method']}',")
                content.append(f"\t\turl=f'{self.domain}/{value['url']}'")

                # Add parameters and data handling
                params = value['params']
                if params:
                    content[-1] += ','
                    args = '{' + ', '.join([f"'{k}': {k}" for k in params.keys()]) + '}'
                    content.append(f"\t\tparams={args}")

                if is_multi:
                    content[-1] += ','
                    content.append('\t\tdata=mp_encoder')
                elif data is not None:
                    content[-1] += ','
                    args = '{' + ', '.join([f"'{k}': {k}" for k in data.keys()]) + '}'
                    if headers.get('Content-Type') == 'application/json':
                        content.append(f"\t\tjson={args}")
                    else:
                        content.append(f"\t\tdata={args}")

                # Add headers
                if headers:
                    content[-1] += ','
                    args = []
                    for k, v in headers.items():
                        if k == 'Content-Type' and is_multi:
                            args.append(f"'{k}': mp_encoder.content_type")
                        else:
                            v = f'self.{k.lower()}' if v in ('str', 'int', 'bool') else f"'{v}'"
                            args.append(f"'{k}': {v}")

                    args = '{' + ', '.join(args) + '}'
                    content.append(f"\t\theaders={args}")

                content[-1] += ')\n'

        return '\n'.join(content)


if __name__ == '__main__':
    # https://square.github.io/retrofit/
    parser = argparse.ArgumentParser(description='Generate Python client methods from Retrofit definitions.')
    parser.add_argument('-d', '--domain', required=False, type=str, metavar='<domain>', default='{self.api}', help='Base domain for API endpoints. (default: {self.api})')
    parser.add_argument('-i', '--input', required=True, type=Path, metavar='<path>', help='Source path (file or directory) containing Java Retrofit definitions.')
    parser.add_argument('-o', '--output', required=False, type=Path, metavar='<path>', default='client.py', help='Output file path for the generated Python client code. (default: client.py)')
    args = parser.parse_args()

    # Initialize Retropy and process input
    retro = Retropy(domain=args.domain)
    retro.load(args.input)

    # Write generated Python client to output file
    result = retro.dump()
    if 'MultipartEncoder' in result:
        print('Required dependency: https://pypi.org/project/requests-toolbelt/')

    path = Path(args.output)
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(result)
    print(f'Output: {path.absolute()}')