def new_writer()

in spectator/writer/new_writer.py [0:0]

• 28 lines of code
• 11 McCabe index (conditional complexity)

def new_writer(location: str) -> WriterUnion:
    """Create a new Writer based on an output location."""

    if location == "none":
        writer = NoopWriter()
    elif location == "memory":
        writer = MemoryWriter()
    elif location == "stderr":
        writer = FileWriter(location, sys.stderr)
    elif location == "stdout":
        writer = FileWriter(location, sys.stdout)
    elif location == "udp":
        # default udp port for spectatord
        location = "udp://127.0.0.1:1234"
        parsed = urlparse(location)
        address = (parsed.hostname, parsed.port)
        writer = UdpWriter(location, address)
    elif location == "unix":
        # default unix domain socket for spectatord
        location = "file:///run/spectatord/spectatord.unix"
        file = open(urlparse(location).path, "a", encoding="utf-8")
        writer = FileWriter(location, file)
    elif location.startswith("file://"):
        file = open(urlparse(location).path, "a", encoding="utf-8")
        writer = FileWriter(location, file)
    elif location.startswith("udp://"):
        parsed = urlparse(location)
        address = (parsed.hostname, parsed.port)
        writer = UdpWriter(location, address)
    else:
        raise ValueError(f"unsupported Writer location: {location}")

    return writer