Source Code Manipulation

So far, whenever we wanted to parse or augment code, we used the AST module – and with good reason, since code is written in a grammar that includes nesting, precedence, and other context-sensitive syntax. To capture it properly, we need a recursive, elaborate tree – but I've also mentioned that code is, at the end of the day, a string.

What will happen if we treat it as such, and use simpler methods like regular expressions, or even plain old search and replace? Feels a bit scary, doesn't it? – and it wouldn't work in all cases, to be sure; but in some it might, and then we'd have saved ourselves a lot of trouble. What's more, we'll see how we can meet it half way: that is, write source code in a way that lends itself to these manipulations, in which case we really can achieve quite a lot without doing anything all that special.

Injecting Logs

To give an example, consider the @log decorator from the previous chapter. It was both simple and effective – the issue, if you remember, was that we could only apply it to the global scope, and even then only dynamically.

But what if we were more intrepid about it? Consider how Python functions are defined: putting aside the seldom used lambda notation, they always start on a new line, and always begin with the def keyword; that's a pretty unique signature. What's more, adding a decoration to one shouldn't disturb the rest of the system – we'll just inject a line before it, in our case @log.

The only thing we need to bear in mind, if we want to support methods and otherwise nested functions, is that any def may be preceded by an arbitrary amount of whitespace, which our decoration line would have to match. That is, given:

<whitespace>def <name>(<parameters>):
<whitespace>    <body>

We want to add:

<whitespace>@log

And that's really not all that hard, and pretty solid, as far as I can tell. The only edge case that comes to mind is multiline strings, which might contain lines starting with whitespace followed by def – but that's a pretty niche case, so I think we can safely ignore it. So then:

import re, sys

def add_log_decorations(source, target):
    code = open(source).read()
    code = f"from log import log\n\n{code}"
    code = re.sub(r"\n( *)def ", r"\n\1@log\n\1def ", code)
    open(target, "w").write(code)
    
if __name__ == "__main__":
    add_log_decorations(sys.argv[1], sys.argv[2])

Assuming we put our decorator in log.py and make sure it's on our import path, this should make it available and apply it to all the function definitions in a given source file. For example, this code:

def add(x, y):
    return x + y
    
class A:
    def f(self):
        def g():
            pass

Can be transformed like so:

$ python add_log_decorations.py example.py logged_example.py
$ cat logged_example.py
from log import log

@log
def add(x, y):
    return x + y
    
class A:
    @log
    def f(self):
        @log
        def g():
            pass

Autodecoration

We can actually generalize and refine this notion. First of all, we can inject an arbitrary decorator; second, we can do so to an entire project's directory at once; and third, we can do so in-place, since the chance of corrupting anything significant is so low, and reverting the changes with basic string operations is equally easy. I'll even add a little cherry on top and show you a cool site-customization trick.

First things first: let's call our script autodecorator.py, make it iterate over entire directories, and add or remove @autodecorator lines depending on an additional command-line argument:

import re, sys, pathlib

def add_decorations(path):
    code = path.read_text()
    code = re.sub(r"\n( *)def ", r"\n\1@autodecorator\n\1def ", code)
    path.write_text(code)

def remove_decorations(path):
    code = path.read_text()
    code = re.sub(r"\n *@autodecorator", "", code)
    path.write_text(code)
    
def main(instruction, root):
    root = pathlib.Path(root)
    for path in root.rglob("*.py"):
        if instruction == "add":
            add_decorations(path)
        if instruction == "remove":
            remove_decorations(path)
            
if __name__ == "__main__":
    main(sys.argv[1], sys.argv[2])

And notice we don't add any import lines: we'll make autodecorator magically available as a builtin with the following command:

$ python autodecorator.py set <file>:<function>

To do this, we'll use the standard site module. Whenever we run python, it imports site, which searches and imports the first sitecustomize.py it finds on its import path; so if we add one in a standard location (like Python's site-packages/ directory), it'll get a chance to execute some code whenever Python runs. This code will add <file>'s directory to sys.path to make sure it's importable, import it, and set <function> as the autodecorator builtin. This way, whatever happens next will have it globally available and properly wired:

# … same as before

def main(instruction, path):
    if instruction in ["add", "remove"]:
        # … same as before
    if instruction == "set":
        path, function = path.split(":")
        set_autodecorator(path, function)

import site

HOOK = """
import builtins, sys
sys.path.append({directory!r})
import {module}
builtins.autodecorator = {module}.{function}
"""

def set_autodecorator(path, function):
    path = pathlib.Path(path).absolute()
    hook = pathlib.Path(site.getsitepackages()[0]) / "sitecustomize.py"
    hook.write_text(HOOK.format(
        directory=str(path.parent),
        module=path.stem,
        function=function,
    ))

Now, what decorator should we try? I think benchmarking is a good candidate: we occasionally need to know how long our functions take, but adding a decorator to each of them manually is tedious, and after we're done we'd have to clean it up. Consider benchmark.py, then:

import atexit, collections, functools, time

benchmarks = collections.defaultdict(list)

def benchmark(f):
    @functools.wraps(f)
    def wrapper(*args, **kwargs):
        started = time.monotonic()
        try:
            return f(*args, **kwargs)
        finally:
            elapsed = time.monotonic() - started
            benchmarks[f.__name__].append(elapsed)
    return wrapper

@atexit.register
def print_benchmarks():
    avgs = [(sum(times) / len(times), name) for name, times in
                                            benchmarks.items()]
    avgs.sort(reverse=True)
    for time, name in avgs:
        print(f"{name}: {time:.02f}s")

The benchmark decorator measures how long a decorated function runs and stores a record in benchmarks, so at the end of the program we can average, sort and print the results. Given a (minimally) elaborate package, p:

# p/f.py
import time

def f():
    time.sleep(0.1)
    
# p/g.py
import time

def g():
    time.sleep(0.2)
    
# p/__init__.py
from .f import f
from .g import g

# p/__main__.py
from . import f, g

def main():
    for _ in range(3):
        f()
        g()
        
main()

We can benchmark its performance as easily as:

$ python autodecorator.py set benchmark.py:benchmark
$ python autodecorator.py add p/
$ python -m p
main: 0.92s
g: 0.2s
f: 0.1s
$ python autodecorator.py remove p/

And profiling is just an example, yeah? This tool lets us imbue our entire codebase with arbitrary additional logic – and do so in a decidedly unobtrusive way, magically configuring it on startup and removing all traces of it after we're done. We can even add an autodecorator.py clear command that deletes sitecustomize.py, but that's left as an exercise to the reader.

Asynchronous Programming

Let me give you an even more daring example; but first, a bit of background. Python 3.5 introduced the async and await keywords, which let us multiplex several execution flows on a single core. The way it works is by running an event loop, which knows to switch from one flow to another whenever some flow gets "stuck" – usually because it's waiting for some slow I/O operation to complete, like fetching rows from a database or communicating with some service over the web.

How do we know when such a blocking operation is happening, so we have time to run other flows until this one's result is ready? Easy – we don't; we expect the flows to tell us explicitly. That's what the await keyword is for: it effectively yields control back to the event loop, suspending the current flow until whatever it's awaiting becomes available and its execution can resume. If the whole thing smells like a bunch of generators, it's because it is: prior to the introduction of these special keywords, such "piecewise" functions, known as coroutines, were implemented by way of generators, using yield from to signal when they were about to block.

That's asynchronous programming in a nutshell: writing coroutines rather than plain functions, accounting for blocking I/O operations by preceding them with await, and deploying the whole thing to some event loop that knows how to switch between them on cue. It's great for performance, as it lets us utilize our compute resources much more economically – the problem with it, except some added complexity that doesn't exist in regular, synchronous programming, is that these two paradigms absolutely do not mix.

You see, if asynchronous code calls a synchronous function that does something "heavy", the entire event loop grinds to a halt. Since it's running on a single core, hogging the CPU prevents other flows from making any progress – but synchronous code is blissfully unaware (and in fact, incapable) of yielding control. It invariably assumes it has the entire core to itself, which is why we need multicore constructs like threads to parallelize it.

And vice versa: if synchronous code attempts to call an asynchronous function, it doesn't work either. Much like with generators, a coroutine is not something we can just invoke, but a gradual process that has to be emulated – and not in a simple loop, mind you, but in a way that performs the delegated I/O, and only ever resumes execution when it's ready. In other words, it necessitates an event loop: one could say coroutines are plugins in a framework that handles the aspect of asynchronous I/O, and plugins usually can't run "on their own".

What's worse, asynchronicity is contagious: if f() calls g(), which calls h(), and h is asynchronous – it means that g must await it, making g asynchronous; and that means f must await it, making f asynchronous as well. And if h is synchronous but does something heavy, it means that if f and g are coroutines, they mustn't call it, lest it clogs their event loop. To conclude: what we write in one paradigm simply cannot be reused in the other – at least, as long as I/O is involved, however deep in the call stack it might be hiding.

Syncification

Still, writing libraries that accommodate both styles is highly desirable: take HTTPX, for example, or SQLAlchemy. On one hand, we'd like to use them synchronously, for when squeezing the most out of our hardware is not that important, so there's no reason to burden ourselves with the added complexity of asynchronous code. On the other hand, we'd also like to make the switch when performance does matter – the problem, as we've established, is that almost no code that touches I/O, however indirectly, can be shared between the two stacks.

How do they do it, then? Well – besides designing around I/O, so at least some code can be reused, they just swallow the bitter pill and implement their functionality twice, once in each paradigm. Naturally, they try to keep the interfaces similar:

>>> import httpx
>>> with httpx.Client() as client:
...     response = client.get("http://www.example.com")
>>> # vs:
>>> async with httpx.AsyncClient() as client:
...     response = await client.get("http://www.example.com")

>>> # or…
>>> from sqlalchemy import create_engine
>>> engine = create_engine("sqlite:///db.sqlite3")
>>> with engine.connect() as conn:
...     rows = conn.execute("SELECT * FROM users")
>>> # vs:
>>> from sqlalchemy.ext.asyncio import create_async_engine
>>> engine = create_async_engine("sqlite+aiosqlite:///db.sqlite3")
>>> async with engine.connect() as conn:
...     rows = await conn.execute("SELECT * FROM users")

So we get to choose which one to use. If, however, we also want to accommodate both styles, we end up in the same boat. Suppose, for example, we're writing the webutils/ library, and we'd like to expose functionality that returns the page size of a given URL in either paradigm. First, we implement:

from httpx import Client

def page_size(url):
    with Client() as client:
        response = client.get(url)
        return len(response.content)

And then again, elsewhere:

from httpx import AsyncClient

async def page_size(url):
    async with AsyncClient() as client:
        response = await client.get(url)
        return len(response.content)

This effectively doubles the code we have to write, debug and maintain – so you have to wonder, can't we just derive one version from the other? Making synchronous code asynchronous is impossible, since we can't automatically guess which operations are heavy enough to warrant an await; but going the other way seems straightforward: simply removing all the asyncs and awaits from the second implementation effectively results in the first.

Here's the idea, then: in a process I call syncification, we take all our asynchronous modules, strip them of these two keywords, and put the resulting copies in a sync/ subpackage. This way, users of our library can do:

>>> from webutils import page_size
>>> size = await page_size("http://www.example.com")

Or, if they prefer the synchronous version:

>>> from webutils.sync import page_size
>>> size = page_size("http://www.example.com")

And we only have to write code once. Here's the script to do so:

import pathlib, re, sys

def syncify(root):
    root = pathlib.Path(root)
    sync = root / "sync"
    sync.mkdir(exist_ok=True)
    for path in root.rglob("*.py"):
        if path.parent.name == "sync":
            continue
        code = path.read_text()
        code = re.sub(r"([aA]sync_?|await) *", "", code)
        sync_path = sync / path.name
        sync_path.write_text(code)
        
if __name__ == "__main__":
    syncify(sys.argv[1])

Running python syncify.py webutils/ will iterate over all the files (except the sync/ directory itself) and copy them into sync/, removing any asyncs and awaits – even if they're capitalized or followed by an underscore. The former adjustment will change classes like AsyncClient into Client; the latter we'll address in a moment.

This works for HTTPX, but what if we were using SQLAlchemy to write a fetch function? The differences there are more pronounced: instead of from sqlalchemy we have from sqlalchemy.ext.asyncio, and instead of create_engine with an sqlite:// prefix, we do create_async_engine with sqlite+aiosqlite://. Of course, we can bruteforce this by adding the necessary substitutions to our script, it's brittle; consider, for instance, how different the code for table creation turns out:

>>> from sqlalchemy import MetaData
>>> metadata = MetaData()
>>> metadata.create_all(engine)
>>> # vs.
>>> async with engine.begin() as conn:
...     await conn.run_sync(metadata.create_all)

Addressing this with a regex is possible, but it ain't pretty. So instead, we'll change our approach, and see if we can meet our solution half way.

Bisynchronous Utilities

Let's start with a single, asynchronous implementation:

from .bisync import async_create_engine

async def fetch(table):
    engine = async_create_engine("sqlite:///db.sqlite3")
    async with engine.connect() as conn:
        rows = await conn.execute("SELECT * FROM users")
        for row in rows:
            yield row

But notice we're not using SQLAlchemy directly; instead, we refactor the contentious bits into a bisync.py module, which will be exempt from our syncification process. This means its code will be shared between the two paradigms – not because it can magically run in both, but because we'll use it to isolate scenarios where they diverge substantially, and provide easily replaceable entrypoints for each. In this case:

import sqlalchemy, sqlalchemy.ext.asyncio

def create_engine(url):
    return sqlalchemy.create_engine(url)

def async_create_engine(url):
    scheme, rest = url.split("://")
    dialect = scheme.split("+")[0]
    driver = "aiosqlite" # or {"sqlite": "aiosqlite", …}[dialect]
    url = f"{dialect}+{driver}://{rest}"
    return sqlalchemy.ext.asyncio.create_async_engine(url)

This way, async_create_engine patches the URL and returns an asynchronous engine, which is what we need; and after syncification, it becomes create_engine (hence the optional _ in our regex), which returns a regular engine – what we need again. More generally, we'd have to write two functions with the same signature per scenario: async_f to handle the asynchronous case and f to handle the synchronous one. When our script strips the async_ bit from the first, it effectively reroutes the code to use the second, paradigm-compatible alternative. Similarly, any code that needs to create tables shouldn't do it directly, but use this instead:

from .bisync import async_create_tables
await async_create_tables(engine, metadata)

Which will become:

from .bisync import create_tables
create_tables(engine, metadata)

Both of which we can implement appropriately in the safe haven of bisync.py:

# … same as before
def create_tables(engine, metadata):
    metadata.create_all(engine)
    
async def async_create_tables(engine, metadata):
    async with engine.begin() as conn:
        await conn.run_sync(metadata.create_all)

Such bisynchronous utilities can be arbitrarily complex: they can even use inherently asynchronous functionality, like tasks, provided that we implement a synchronous equivalent (e.g. with threads). Suppose, for example, that we'd want our page_size function to accept multiple URLs and calculate their sizes concurrently. Doing something like:

import asyncio
from httpx import AsyncClient

async def page_size(*urls):
    async with AsyncClient() as client:
        tasks = [asyncio.create_task(client.get(url)) for url in urls]
        responses = asyncio.gather(*tasks)
        return [len(response.content) for response in responses]

Is out of the question, since there's no reasonable way to syncify this code via search-and-replace (or even AST). Doing this, however:

from httpx import AsyncClient
from .bisync import async_parallelize

async def page_size(*urls):
    async with AsyncClient() as client:
        responses = await async_parallelize(client.get, urls)
        return [len(response.content) for response in responses]

Would work, since the syncified version:

from httpx import Client
from .bisync import parallelize

def page_size(*urls):
    with Client() as client:
        responses = parallelize(client.get, urls)
        return [len(response.content) for response in responses]

Makes just as much sense, provided we implement async_parallelize and parallelize accordingly:

import asyncio, concurrent.futures

def async_parallelize(f, args):
    tasks = [asyncio.create_task(f(arg)) for arg in args]
    return asyncio.gather(*tasks)
    
def parallelize(f, args):
    with concurrent.futures.ThreadPoolExecutor() as executor:
        return list(executor.map(f, args))

Import Normalization

One issue we have to address with syncification is that it screws up our project structure. Right now, it simply takes all the Python files within our package and "flattens" them into a sync subpackage – so modules from different directories that have a similar name will overwrite each other, for one. That actually doesn't bother me too much, since we can decide to call our modules by unique names (meeting our solution half way, yeah?); a much more acute problem is that when one module depends on another, moving them around breaks imports.

With relative imports it's quite obvious: something like from ...x import no longer makes sense from inside <package>.sync.<module>. Absolute imports will still compile, but they will import the wrong thing, going for the asynchronous implementation instead of the synchronous equivalent. So either way, we need to fix this – luckily, a simple search-and-replace will once again do the trick.

Both scenarios can be addressed, but personally I prefer relative imports, so I'll do those. What we need is to catch any line that has the form from .… and point it to the syncified version instead. Since all the synchronous modules are sitting next to each other inside sync/, it means from .x.y.z should become from .z, from ..x.y should become from .y, and from ...x should become from .x – all of which can be done by a single regex:

def syncify(root):
    # … same as before
    code = path.read_text()
    code = re.sub(r"from \.\S*?([a-zA-Z_][a-zA-Z0-9_]*) ", r"from .\1 ", code)
    # … same as before

The only exception is from .bisync, which, assuming it sits at the root of the package, should become from ..bisync – but this doesn't even require a regular expression:

def syncify(root):
    # … same as before
    code = path.read_text()
    code = re.sub(r"from \.\S*?([a-zA-Z_][a-zA-Z0-9_]*) ", r"from .\1 ", code)
    code = code.replace("from .bisync ", "from ..bisync ")
    # … same as before

And this takes care of most import-related problems, if not all. Some issues might crop up, of course, but nothing major: I'll tell you about one I had, just to wrap up nicely. When I work this way, I also duplicate my tests, just to make sure syncification didn't break anything (the probability of both the code and its test mutating in a wrong-yet-coordinated way strikes me as pretty slim). To test asynchronous code, however, we need the pytest-asyncio extension, and a special marker to make Pytest consider coroutines as tests:

import pytest
import pytest_asyncio # might be needed, e.g. for @pytest_asyncio.fixture

from webutils import page_size

pytestmark = pytest.mark.asyncio

async def test_page_size():
    assert await page_size("http://www.example.com") == 1256

To syncify this, we need to remove the pytest_asyncio import and marker; but then, if I don't use pytest anywhere else in the code, the linter complains that import pytest is unused, so we might need to remove it, too. And of course, we need to change from webutils import to from webutils.sync import, so as to wire the tests to our synchronous code. Cleaning this up is a bit particular, but aside from that I can't think of any special handling I had to do in my projects, so this should really cover most our bases:

def syncify(root):
    # … same as before
    tests_root = root.parent / "tests"
    async_tests = tests_root / "async"
    sync_tests = tests_root / "sync"
    sync_tests.mkdir(exist_ok=True)
    for path in async_tests.rglob("*.py"):
        code = path.read_text()
        code = re.sub(f"from {root.name}", f"from {root.name}.sync", code)
        if "import pytest\n" in code:
            code = code.replace("import pytest_asyncio\n", "")
        code = code.replace("pytest_asyncio", "pytest")
        code = re.sub(r"\n*pytestmark = pytest.mark.asyncio", "", code)
        if "import pytest" in code and code.count("pytest") == 1:
            code = code.replace("import pytest\n", "")
        code = re.sub(r"([aA]sync_?|await) *", "", code)
        sync_path = sync_tests / path.relative_to(async_tests)
        sync_path.parent.mkdir(exist_ok=True, parents=True)
        sync_path.write_text(code)

Circular Imports

Imports, as you can see, are a bit of a hassle; not only when we're engaged in mad metaprogramming, but in perfectly reasonable code as well. So before we close, let me take a short detour from manipulating source into the import mechanics themselves – it will pay off, I promise, with a finale that brings us right back to treating code as a string. Consider a simple fileutils/ package, which has a Directory class in one module and a File class in another:

# fileutils/directory.py
from .file import File
class Directory:
    def get_files(self) -> list[File]:
        …
        
# fileutils/file.py
from .directory import Directory
class File:
    @property
    def directory(self) -> Directory:
        …

This code doesn't work, on account of a circular import: directory.py attempts to import file.py, which in turn attempts to import directory.py, at which point the interpreter crashes with an ImportError. This, despite the fact that a circular dependencies between files and directories is perfectly normal, not due to some contrived design on our part.

Python's typing library comes equipped to handle such cases with the special TYPE_CHECKING flag, which lets us place otherwise problematic dependencies in an if-statement that will be False at runtime:

# fileutils/directory.py
from typing import TYPE_CHECKING
if TYPE_CHECKING:
    from .file import File
# … same as before

# fileutils/file.py
from typing import TYPE_CHECKING
if TYPE_CHECKING:
    from .directory import Directory
# … same as before

This way, our code no longer crashes – at least, not immediately. The problem is that our particular circular dependency exists in runtime as well; besides their annotations, the methods in question will have to instantiate the actual object in their bodies, e.g.:

class File:
    @property
    def directory(self) -> Directory:
        return Directory(…)

At which point our program raises a NameError, since Directory is never defined when the TYPE_CHECKING clause is skipped. Luckily, there are several ways to bypass this, too – we can, for instance, place the problematic imports at the bottom of the file instead of its top:

# fileutils/directory.py
from __future__ import annotations
class Directory:
    def get_files(self) -> list[File]:
        …
from .file import File

# fileutils/file.py
from __future__ import annotations
class File:
    @property
    def directory(self) -> Directory:
        …
from .directory import Directory

We need from __future__ import annotations to defer annotation resolution, but otherwise this works, thanks to Python's support of partial initialization: by the time directory.py attempts to import file.py, the Directory class is already defined, so when file.py attempts to import it from directory.py it actually works, despite the fact directory.py as a whole is still being evaluated.

Still, putting imports at the bottom feel a bit awkward; Google's coding standards recommend doing this instead:

# fileutils/file.py
from typing import TYPE_CHECKING
from . import directory
if TYPE_CHECKING:
    from .directory import Directory
    
class File:
    @property
    def directory(self) -> Directory:
        return directory.Directory(…)

By importing the module instead of the class, we're once again relying on partial initialization: the statement succeeds even though it's still being evaluated, and by the time it's actually used, its objects are already fully defined. Using it for annotations, however, would access it too early, so we need to combine this with the TYPE_CHECKING method; I can't say the result looks very elegant either.

What's worse, as our project gets big, we might have subpackages; and if we want to expose their internals by importing them in the subpackage __init__.py, any circularity between them becomes untenable – so I'd like to show you one more trick; not exactly about source code manipulation per se, but a program that reads its own source code to recalibrate itself – given our chapter's subject and progression, I thought it might fit.

Lazy Imports

Imagine we could just do import <package> everywhere, and accessing any name we wish to expose would locate and evaluate it on demand. This not only solves circularity problems, but makes importing blazing fast, since various parts of our code are evaluated only when they're needed. We'd still need TYPE_CHECKING for annotations, so as not to resolve their classes until they're actually required, but the plan is to have everything centralized in a single, automatically-managed point:

from typing import TYPE_CHECKING
import fileutils
if TYPE_CHECKING:
    from fileutils import Directory
    
class File:
    @property
    def directory(self) -> Directory:
        return fileutils.Directory(…)

Doesn't look too different from the previous version, but if we have multiple imports, or they're located in inter-dependent subpackages, trust me – this is much simpler. Plus, we no longer need to choose between relative and absolute styles, or care about where each part is located, since we only ever use the root package.

That is, except in the root __init__.py, where we manage all the names that need to be exposed – either as part of our public API, or because we need them elsewhere in our package. It might feel wrong to mix the two, effectively exporting internals – but in my experience, especially when typing is used, accessing every corner of the package becomes necessary, and it's quite annoying to go fishing for definitions in obscure modules, as is ending up with a long list of imports from all over the place. Here's a snippet of my imports when working with the OpenAI library:

from openai.types.chat.chat_completion import Choice
from openai.types.chat.chat_completion_chunk import Choice as StreamChoice
from openai.types.chat.chat_completion_content_part_image_param import ImageURL
from openai.types.chat.chat_completion_content_part_param import File
from openai.types.chat.chat_completion_message_tool_call_param import Function

Yeah, not very fun.

Anyway, how do write our __init__.py? Since we're going to expose its attributes dynamically, it will initially have no attributes at all, which means that if we want to support things like typing and autocompletion, we'd have to bring everything forth in the TYPE_CHECKING clause:

from typing import TYPE_CHECKING

if TYPE_CHECKING:
    from .file import File
    from .directory import Directory

Once we do that, however, we can reuse these definitions – presumably, there'd be a lot, not just the two of our toy example. When we attempt to load e.g. File, our lazy loader will parse this __init__.py to match the names to their sources, then import the relevant source and extract the name, caching the result for good measure (I'd also cache the imported module, if Python's import system wouldn't do that for us anyway). Let's start, then, by defining an importer class in the same __init__.py:

# … same as before

class __Importer:

    def __init__(self):
        self.sources = {}
        self.objects = {}
        self.re = __import__("re")
        self.importlib = __import__("importlib")

And notice I'm trying to avoid polluting the package namespace, hence the __ prefix and the internal imports of a couple libraries we'll need in a moment. To load the sources, as discussed, we do:

class __Importer:
    # … same as before
    def load_sources(self):
        code = open(__file__).read()
        for source, names in self.re.findall(r"from (.*?) import (.*)", code):
            if source.startswith("."):
                for name in names.split(","):
                    self.sources[name.strip()] = source

That is, it reads and parses its own source code, extracting relative import lines to end up with File mapped to .file and Directory to .directory. Then:

class __Importer:
    # … same as before
    def __call__(self, name):
        if not self.sources:
            self.load_sources()
        if name not in self.objects:
            if name not in self.sources:
                raise AttributeError(f"module {__name__!r} has no "
                                    f"attribute {name!r}")
            source = self.sources[name]
            module = self.importlib.import_module(source, __package__)
            self.objects[name] = getattr(module, name)
        return self.objects[name]

Which, as promised, loads and extracts the relevant file. Here, I'm taking advantage of the fact that importlib.import_module knows how to resolve relative locations (like .file) given a package; and since we're operating from within the __init__.py of said package, __package__ it is. Last but not least – all of this happens in __call__, which means our importer object is callable, and all that's left is to bind as our package's name resolver:

# … same as before
__getattr__ = __Importer()

Just like the __getattr__ method is invoked to dynamically resolve attributes, so is the global __getattr__ of a module called on names that aren't explicitly defined in its scope. The result:

>>> import fileutils
>>> fileutils.Directory
<class 'fileutils.directory.Directory'>
>>> fileutils.File
<class 'fileutils.file.File'>

All in all, not bad for some 20 lines – and never having to worry about imports again. Like I said, we don't exactly manipulate source code in this example, but we do parse it to determine which name comes whence; the lesson I'm trying to impart is not so much this or that method, but the fact that code is, ultimately, a string – so sometimes, thinking of it as such comes in handy.