Configuration¶
ADB Binary Path¶
By default, adbflow looks for adb on your system PATH. To specify a custom path:
from adbflow import ADB
adb = ADB(adb_path="/opt/android-sdk/platform-tools/adb")
device = await adb.device_async()
Transport Layer¶
All ADB communication goes through the SubprocessTransport abstraction. You typically don't interact with it directly, but it's available if needed:
# Access the transport from a device
transport = device.transport
# Execute a raw ADB command
result = await transport.execute(["devices", "-l"])
print(result.stdout)
# Execute a shell command on a specific device
result = await transport.execute_shell("getprop ro.product.model", serial="emulator-5554")
# Stream output line by line
async for line in transport.stream_lines(["logcat"], serial=device.serial):
print(line)
Async Patterns¶
adbflow is async-first. All I/O methods have an _async suffix and return coroutines.
Running async code¶
import asyncio
from adbflow import ADB
async def main():
adb = ADB()
device = await adb.device_async()
model = await device.shell_async("getprop ro.product.model")
print(model)
asyncio.run(main())
Sync wrappers¶
Every async method has a sync counterpart (without _async) that calls asyncio.run() internally:
from adbflow import ADB
adb = ADB()
device = adb.device()
model = device.shell("getprop ro.product.model")
Warning
Sync wrappers create a new event loop each call. Don't mix sync wrappers with an existing async context — use the _async methods directly instead.
Parallel operations¶
Use asyncio.gather to run independent operations concurrently:
async def parallel_example(device):
model, version, battery = await asyncio.gather(
device.info.model_async(),
device.info.android_version_async(),
device.info.battery_async(),
)
Timeouts¶
Most operations accept a timeout parameter (in seconds):
# Shell command with timeout
result = await device.shell_async("long-running-command", timeout=30.0)
# Wait for UI element with timeout
element = await device.ui.wait_for_async(
Selector().text("Done"),
timeout=15.0
)
Operations that exceed their timeout raise ADBTimeoutError.
Device Properties¶
Device properties are cached for performance. Invalidate the cache if you need fresh data:
# Cached after first call
props = await device.properties.get_all_async()
# Force refresh
props = await device.properties.get_all_async(refresh=True)
# Or invalidate manually
device.properties.invalidate_cache()
UI Hierarchy Cache¶
The UI hierarchy dump is cached for 2 seconds to avoid expensive re-dumps during rapid queries. The cache is automatically invalidated after any gesture or interaction. Force a fresh dump with:
Or invalidate manually: