Building an Appium Framework in Python

Schema

{
  "<device_type>": {          // emulator | real_device | browserstack
    "default": "android",
    "android": { ...caps, "appiumServerUrl": "..." },
    "ios":     { ...caps, "appiumServerUrl": "..." }
  }
}

Profile strings

• emulator → emulator[default]
• emulator.ios → emulator.ios
• browserstack → browserstack[default]

Example (abbreviated)

{
  "emulator": {
    "default": "android",
    "android": {
      "platformName": "Android",
      "appium:deviceName": "emulator-5554",
      "appium:app": "./path/to/your/app.apk",
      "appium:automationName": "UiAutomator2",
      "appiumServerUrl": "http://127.0.0.1:4723"
    },
    "ios": {
      "platformName": "iOS",
      "appium:deviceName": "iPhone 17",
      "appium:app": "./path/to/your/app.app",
      "appium:automationName": "XCUITest",
      "appiumServerUrl": "http://127.0.0.1:4723"
    }
  },
  "browserstack": {
    "default": "android",
    "android": {
      "platformName": "android",
      "appium:app": "bs://your-app-id",
      "bstack:options": {
        "userName": "$BROWSERSTACK_USERNAME",
        "accessKey": "$BROWSERSTACK_ACCESS_KEY"
      },
      "appiumServerUrl": "https://hub.browserstack.com"
    }
  }
}

Rules worth remembering:

• Appium caps use the appium: prefix (W3C / Appium 2)
• BrowserStack options live under bstack:options
• appiumServerUrl is popped out by ConfigReader — it is not a W3C capability
• Relative appium:app paths resolve against the config file directory
• Strings like $BROWSERSTACK_USERNAME resolve from the process environment

1. ConfigReader

Loads JSON, resolves profile / env / app path, returns (capabilities, server_url).

import json
import os
import re

from glom import PathAccessError, glom

_ENV_VAR_PATTERN = re.compile(r"^\$([A-Za-z_][A-Za-z0-9_]*)$")
_CONFIG_ENV_VAR = "APPIUM_BDD_CONFIG"


class ConfigReader:
    def __init__(self, config_file_path=None):
        if config_file_path is None:
            config_file_path = os.environ.get(_CONFIG_ENV_VAR, "config.json")

        self.config_file_path = os.path.abspath(config_file_path)
        self._config_dir = os.path.dirname(self.config_file_path)

        with open(self.config_file_path, "r") as file:
            self.config_data = json.load(file)

    @staticmethod
    def _resolve_env_placeholders(value):
        if isinstance(value, str):
            match = _ENV_VAR_PATTERN.match(value)
            if match:
                env_var_name = match.group(1)
                env_value = os.environ.get(env_var_name)
                if env_value is None:
                    raise ValueError(
                        f"Environment variable '{env_var_name}' referenced in config.json "
                        "is not set."
                    )
                return env_value
            return value
        if isinstance(value, dict):
            return {
                key: ConfigReader._resolve_env_placeholders(val)
                for key, val in value.items()
            }
        if isinstance(value, list):
            return [ConfigReader._resolve_env_placeholders(item) for item in value]
        return value

    def _resolve_app_path(self, capabilities):
        app_path = capabilities.get("appium:app")
        if not app_path or not isinstance(app_path, str):
            return capabilities
        if app_path.startswith("bs://") or os.path.isabs(app_path):
            return capabilities
        capabilities["appium:app"] = os.path.normpath(
            os.path.join(self._config_dir, app_path)
        )
        return capabilities

    def _resolve_profile(self, profile):
        device_type, _, variant = profile.partition(".")

        try:
            group = glom(self.config_data, device_type)
            return glom(group, variant or group["default"])
        except (PathAccessError, KeyError) as exc:
            if device_type not in self.config_data:
                available = ", ".join(sorted(self.config_data))
                raise ValueError(
                    f"Device type '{device_type}' not found in config file. "
                    f"Available: {available}"
                ) from exc

            variants = ", ".join(
                sorted(k for k in self.config_data[device_type] if k != "default")
            )
            raise ValueError(
                f"Profile '{profile}' not found. "
                f"Available variants under '{device_type}': {variants}"
            ) from exc

    def get_capabilities(self, profile="emulator"):
        capabilities = self._resolve_env_placeholders(
            self._resolve_profile(profile).copy()
        )
        capabilities = self._resolve_app_path(capabilities)
        appium_server_url = capabilities.pop("appiumServerUrl", None)
        return capabilities, appium_server_url

2. Platform drivers

Appium Python Client v5 expects W3C Options — not the old desired_capabilities dict.

# core/driver/platforms/android_driver.py
from appium import webdriver
from appium.options.android import UiAutomator2Options


class AndroidDriver:
    def __init__(self, capabilities, appium_server_url):
        self.capabilities = capabilities
        self.appium_server_url = appium_server_url

    def create_driver(self):
        options = UiAutomator2Options()
        options.load_capabilities(self.capabilities)
        return webdriver.Remote(self.appium_server_url, options=options)

# core/driver/platforms/ios_driver.py
from appium import webdriver
from appium.options.ios import XCUITestOptions


class IOSDriver:
    def __init__(self, capabilities, appium_server_url):
        self.capabilities = capabilities
        self.appium_server_url = appium_server_url

    def create_driver(self):
        options = XCUITestOptions()
        options.load_capabilities(self.capabilities)
        return webdriver.Remote(self.appium_server_url, options=options)

3. DriverFactory + DriverManager

The factory branches on platformName only. Emulator vs BrowserStack is already encoded in caps + server URL.

from core.configuration.config_reader import ConfigReader
from core.driver.platforms.android_driver import AndroidDriver
from core.driver.platforms.ios_driver import IOSDriver


class DriverFactory:
    def __init__(self, profile):
        self.profile = profile
        self.config_reader = ConfigReader()

    def create_driver(self):
        capabilities, appium_server_url = self.config_reader.get_capabilities(
            self.profile
        )
        platform_name = capabilities.get("platformName").lower()

        if platform_name == "android":
            return AndroidDriver(capabilities, appium_server_url).create_driver()
        elif platform_name == "ios":
            return IOSDriver(capabilities, appium_server_url).create_driver()
        else:
            raise ValueError(f"Unsupported platform: {platform_name}")

from core.driver.driver_factory import DriverFactory


class DriverManager:
    _driver_instance = None

    @classmethod
    def initialize_driver(cls, profile="emulator"):
        if cls._driver_instance is None:
            factory = DriverFactory(profile)
            cls._driver_instance = factory.create_driver()
        return cls._driver_instance

    @classmethod
    def quit_driver(cls):
        if cls._driver_instance:
            cls._driver_instance.quit()
            cls._driver_instance = None

Hooks call initialize_driver / quit_driver. For parallel runs, swap the singleton for thread-local storage — the rest of the design stays the same.

4. DriverHelper

Session-level utilities only (alerts, context switches). No screen-specific logic.

from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC
from selenium.common.exceptions import NoAlertPresentException

from core.driver.driver_manager import DriverManager


class DriverHelper:
    @staticmethod
    def is_alert_present():
        try:
            WebDriverWait(DriverManager._driver_instance, 3).until(
                EC.alert_is_present()
            )
            return True
        except NoAlertPresentException:
            return False

    @classmethod
    def accept_alert(cls):
        if DriverHelper.is_alert_present():
            alert = DriverManager._driver_instance.switch_to.alert
            alert.accept()

5. Element

One interaction surface. Supports a shared locator or per-platform locators. Gestures use W3C mobile:* commands.

from selenium.webdriver.remote.webelement import WebElement
from core.driver.driver_manager import DriverManager
from selenium.webdriver.common.by import By
from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC


class Element:
    def __init__(
        self,
        locator_type: By | None = None,
        locator_value: str | None = None,
        *,
        android: tuple[By, str] | None = None,
        ios: tuple[By, str] | None = None,
    ):
        if android is not None or ios is not None:
            self._platform_locators = {"android": android, "ios": ios}
            self.locator_type = None
            self.locator_value = None
        else:
            if locator_type is None or locator_value is None:
                raise ValueError(
                    "Provide either (locator_type, locator_value) "
                    "or android=/ios= platform locators."
                )
            self._platform_locators = None
            self.locator_type = locator_type
            self.locator_value = locator_value
        self._element: WebElement | None = None

    def _resolve_locator(self) -> tuple[By, str]:
        if self._platform_locators is None:
            return self.locator_type, self.locator_value

        platform = self.driver.capabilities.get("platformName", "").lower()
        locator = self._platform_locators.get(platform)
        if locator is None:
            raise ValueError(f"No locator defined for platform '{platform}'.")
        return locator

    def __str__(self):
        if self._platform_locators is not None:
            return f"Element(platform_locators={self._platform_locators})"
        return (
            f"Element(locator_type={self.locator_type}, "
            f"locator_value={self.locator_value})"
        )

    def __repr__(self):
        if self._platform_locators is not None:
            return f"Element(platform_locators={self._platform_locators!r})"
        return (
            f"Element(locator_type={self.locator_type!r}, "
            f"locator_value={self.locator_value!r})"
        )

    @property
    def driver(self):
        if not DriverManager._driver_instance:
            raise ValueError("Driver not initialized")
        return DriverManager._driver_instance

    def wait_for_element_present(self, timeout: int = 10) -> WebElement:
        return WebDriverWait(self.driver, timeout).until(
            EC.presence_of_element_located(self._resolve_locator())
        )

    def wait_for_element_visible(self, timeout: int = 10) -> WebElement:
        return WebDriverWait(self.driver, timeout).until(
            EC.visibility_of_element_located(self._resolve_locator())
        )

    def wait_for_element_clickable(self, timeout: int = 10) -> WebElement:
        return WebDriverWait(self.driver, timeout).until(
            EC.element_to_be_clickable(self._resolve_locator())
        )

    def find_element(self) -> WebElement:
        if not self._element:
            self._element = self.wait_for_element_present()
        return self._element

    def tap(self) -> None:
        element = self.wait_for_element_clickable()
        element.click()

    def type(self, text: str) -> None:
        element = self.wait_for_element_clickable()
        element.send_keys(text)

    def drag_and_drop(self, destination_element: WebElement) -> None:
        origin_el = self.find_element()
        origin_rect = origin_el.rect
        dest_rect = destination_element.rect

        start_x = origin_rect["x"] + origin_rect["width"] // 2
        start_y = origin_rect["y"] + origin_rect["height"] // 2
        end_x = dest_rect["x"] + dest_rect["width"] // 2
        end_y = dest_rect["y"] + dest_rect["height"] // 2

        platform = self.driver.capabilities.get("platformName", "").lower()
        if platform == "ios":
            self.driver.execute_script(
                "mobile: dragFromToForDuration",
                {
                    "duration": 1.0,
                    "fromX": start_x,
                    "fromY": start_y,
                    "toX": end_x,
                    "toY": end_y,
                },
            )
        else:
            self.driver.execute_script(
                "mobile: dragGesture",
                {
                    "elementId": origin_el.id,
                    "endX": end_x,
                    "endY": end_y,
                },
            )

    def swipe(self, end_x: int, end_y: int, duration: int = 0) -> None:
        start_rect = self.get_element_rect()
        start_x = start_rect["x"] + start_rect["width"] // 2
        start_y = start_rect["y"] + start_rect["height"] // 2

        direction = "up" if end_y < start_y else "down"
        self.driver.execute_script(
            "mobile: swipeGesture",
            {
                "left": start_x,
                "top": start_y,
                "width": 1,
                "height": 1,
                "direction": direction,
                "percent": 1.0,
            },
        )

    def get_text(self) -> str:
        element = self.wait_for_element_visible()
        return element.text

    def get_element_rect(self) -> dict:
        element = self.find_element()
        return {
            "x": element.rect["x"],
            "y": element.rect["y"],
            "width": element.rect["width"],
            "height": element.rect["height"],
        }

    def perform_swipe_up(self, duration=300):
        rect = self.driver.get_window_rect()
        self.driver.execute_script(
            "mobile: swipeGesture",
            {
                "left": rect["x"],
                "top": rect["y"],
                "width": rect["width"],
                "height": rect["height"],
                "direction": "up",
                "percent": 0.75,
            },
        )

    def scroll_to_element(self, max_scrolls=5, scroll_duration=300):
        for _ in range(max_scrolls):
            if self.is_displayed():
                return
            self.perform_swipe_up(duration=scroll_duration)

    def is_displayed(self) -> bool:
        try:
            return self.find_element().is_displayed()
        except Exception:
            return False

6. Locator

Declare Android + iOS tuples (with {param} templates) and resolve to Element.

import re
from typing import Any

from appium.webdriver.common.appiumby import AppiumBy
from selenium.webdriver.common.by import By

from core.element.element import Element

LocatorTuple = tuple[By, str]


class Locator:
    def __init__(self, *, android: LocatorTuple, ios: LocatorTuple):
        self.android = android
        self.ios = ios
        self._params = _template_params(android[1], ios[1])

    @classmethod
    def id(cls, name: str) -> Element:
        spec = (AppiumBy.ACCESSIBILITY_ID, name)
        return cls(android=spec, ios=spec).resolve()

    @classmethod
    def text(cls, text: str) -> Element:
        return cls(
            android=(AppiumBy.ANDROID_UIAUTOMATOR, f'new UiSelector().text("{text}")'),
            ios=(AppiumBy.IOS_PREDICATE, f'name == "{text}"'),
        ).resolve()

    def resolve(self, *args: Any, **kwargs: Any) -> Element:
        params = _bind_params(
            self._params, args, kwargs, name=getattr(self, "_name", "Locator")
        )
        return Element(
            android=_format(self.android, params),
            ios=_format(self.ios, params),
        )

    def __set_name__(self, owner, name):
        self._name = name

    def __get__(self, obj, objtype=None):
        return self.resolve


def _template_params(*values: str) -> list[str]:
    names: list[str] = []
    for value in values:
        for match in re.findall(r"\{(\w+)\}", value):
            if match not in names:
                names.append(match)
    return names


def _bind_params(
    param_names: list[str],
    args: tuple[Any, ...],
    kwargs: dict[str, Any],
    *,
    name: str,
) -> dict[str, Any]:
    if not param_names:
        return dict(kwargs)

    params = dict(kwargs)
    for index, arg in enumerate(args):
        if index >= len(param_names):
            raise TypeError(
                f"{name}() takes {len(param_names)} argument(s), "
                f"got {len(args)} positional."
            )
        params[param_names[index]] = arg

    missing = [key for key in param_names if key not in params]
    if missing:
        raise TypeError(f"{name}() missing argument(s): {', '.join(missing)}")

    return params


def _format(spec: LocatorTuple, params: dict[str, Any]) -> LocatorTuple:
    by, value = spec
    return by, value.format(**params) if params else value

Class attributes become callables via __get__:

class DragPage:
    _box_drop = Locator(
        android=(AppiumBy.ACCESSIBILITY_ID, "drop-{order}"),
        ios=(AppiumBy.IOS_PREDICATE, 'name == "drop-{order}"'),
    )

# self._box_drop("l2")  → Element

7. Strategies + annotations

Annotation keywords map to AppiumBy. Add a strategy = one dict entry.

from appium.webdriver.common.appiumby import AppiumBy
from selenium.webdriver.common.by import By

STRATEGIES: dict[str, By] = {
    "accessibility": AppiumBy.ACCESSIBILITY_ID,
    "accessibility_id": AppiumBy.ACCESSIBILITY_ID,
    "id": AppiumBy.ACCESSIBILITY_ID,
    "xpath": AppiumBy.XPATH,
    "uiautomator": AppiumBy.ANDROID_UIAUTOMATOR,
    "android_uiautomator": AppiumBy.ANDROID_UIAUTOMATOR,
    "predicate": AppiumBy.IOS_PREDICATE,
    "ios_predicate": AppiumBy.IOS_PREDICATE,
    "class_chain": AppiumBy.IOS_CLASS_CHAIN,
    "ios_class_chain": AppiumBy.IOS_CLASS_CHAIN,
}


def strategy_tuple(**kwargs: str) -> tuple[By, str]:
    if len(kwargs) != 1:
        raise ValueError(f"Expected exactly one strategy keyword, got: {kwargs}")

    name, value = next(iter(kwargs.items()))
    by = STRATEGIES.get(name)
    if by is None:
        supported = ", ".join(sorted(STRATEGIES))
        raise ValueError(f"Unknown strategy '{name}'. Supported: {supported}")

    return by, value

import inspect
from collections.abc import Callable
from functools import wraps
from typing import Any

from core.element.locator import Locator
from core.element.strategies import strategy_tuple

_FIND_BY = "__find_by__"


def AndroidFindBy(**kwargs: str) -> Callable:
    return _platform_find_by("android", kwargs)


def iOSFindBy(**kwargs: str) -> Callable:
    return _platform_find_by("ios", kwargs)


def _platform_find_by(platform: str, kwargs: dict[str, str]) -> Callable:
    if not kwargs:
        raise ValueError(f"@{platform}FindBy requires a locator strategy keyword.")

    spec = strategy_tuple(**kwargs)

    def decorator(func: Callable) -> Callable:
        if getattr(func, "_is_find_by", False):
            target = func
        else:
            target = _wrap(func)

        pairs: dict[str, tuple] = getattr(target, _FIND_BY, {})
        pairs[platform] = spec
        setattr(target, _FIND_BY, pairs)
        return target

    return decorator


def _wrap(func: Callable) -> Callable:
    @wraps(func)
    def wrapper(self, *args, **kwargs):
        pairs: dict[str, tuple] = getattr(wrapper, _FIND_BY)
        if "android" not in pairs or "ios" not in pairs:
            raise ValueError(
                f"{func.__name__} needs both @AndroidFindBy and @iOSFindBy."
            )

        params = _method_params(func, self, args, kwargs)
        return Locator(android=pairs["android"], ios=pairs["ios"]).resolve(**params)

    wrapper._is_find_by = True
    return wrapper


def _method_params(
    func: Callable, self: Any, args: tuple, kwargs: dict
) -> dict[str, Any]:
    bound = inspect.signature(func).bind(self, *args, **kwargs)
    bound.apply_defaults()
    return {key: value for key, value in bound.arguments.items() if key != "self"}

Both platforms are required. Method params (e.g. order) feed {order} in locator strings.

8. AllureReport + JsonUtil

import os
import allure
from allure_commons.types import AttachmentType
from behave import fixture


class AllureReport:
    def __init__(self):
        self.results_dir = "allure-results"
        if not os.path.exists(self.results_dir):
            os.makedirs(self.results_dir)

    @fixture
    def setup_allure_report(self, context):
        context.results_dir = self.results_dir
        context.allure = allure
        yield

    def log_step(self, step_name, status, exception=None):
        with allure.step(step_name):
            if status == "failed":
                allure.attach(
                    str(exception),
                    name="Exception Details",
                    attachment_type=AttachmentType.TEXT,
                )
            allure.step(f"Step Status: {status}")

    def attach_screenshot(self, context, name="Screenshot"):
        if hasattr(context, "driver"):
            screenshot = context.driver.get_screenshot_as_png()
            allure.attach(
                screenshot, name=name, attachment_type=AttachmentType.PNG
            )

    def attach_exception(self, step):
        if step.status == "failed":
            allure.attach(
                str(step.exception),
                name="Exception",
                attachment_type=allure.attachment_type.TEXT,
            )

Wire Allure from Behave hooks — not from page objects.

import json
from typing import Type, TypeVar, Dict

T = TypeVar("T")


class JsonUtil:
    @staticmethod
    def parse_json_to_dto(json_path: str, DtoClass: Type[T]) -> Dict[str, T]:
        with open(json_path, "r") as file:
            data = json.load(file)
        return {key: DtoClass(**value) for key, value in data.items()}