pymediawikidocker API Documentation

config

Created on 2023-04-06

@author: wf

Host

Host name getter

Source code in mwdocker/config.py

class Host:
    """
    Host name getter
    """

    @classmethod
    def get_default_host(cls) -> str:
        """
        Get the default host as a usable hostname or IP,
        never returning reverse-DNS PTRs and avoiding localhost which
        might cause to try socket access instead of proper host access
        """
        host = socket.getfqdn()

        # work around https://github.com/python/cpython/issues/79345
        if host == (
            "1.0.0.0.0.0.0.0.0.0.0.0.0.0.0.0.0.0.0.0.0.0.0.0.0.0.0.0.0.0.0.0.ip6.arpa"
        ):
            host = "127.0.0.1"

        elif host.endswith(".in-addr.arpa"):
            host = "127.0.0.1"

        elif host.endswith(".ip6.arpa"):
            host = "::1"

        elif host == "localhost":
            host = "127.0.0.1"

        return host

get_default_host() classmethod

Get the default host as a usable hostname or IP, never returning reverse-DNS PTRs and avoiding localhost which might cause to try socket access instead of proper host access

Source code in mwdocker/config.py

@classmethod
def get_default_host(cls) -> str:
    """
    Get the default host as a usable hostname or IP,
    never returning reverse-DNS PTRs and avoiding localhost which
    might cause to try socket access instead of proper host access
    """
    host = socket.getfqdn()

    # work around https://github.com/python/cpython/issues/79345
    if host == (
        "1.0.0.0.0.0.0.0.0.0.0.0.0.0.0.0.0.0.0.0.0.0.0.0.0.0.0.0.0.0.0.0.ip6.arpa"
    ):
        host = "127.0.0.1"

    elif host.endswith(".in-addr.arpa"):
        host = "127.0.0.1"

    elif host.endswith(".ip6.arpa"):
        host = "::1"

    elif host == "localhost":
        host = "127.0.0.1"

    return host

MwClusterConfig dataclass

Bases: MwConfig

MediaWiki Cluster configuration for multiple wikis

Source code in mwdocker/config.py

@dataclass
class MwClusterConfig(MwConfig):
    """
    MediaWiki Cluster configuration for multiple wikis
    """

    versions: Optional[List[str]] = field(
        default_factory=lambda: ["1.35.13", "1.39.17", "1.43.6", "1.44.3", "1.45.1"]
    )
    base_port: int = 9080

    def addArgs(self, parser):
        """
        add my arguments to the given parser
        """
        super().addArgs(parser)
        parser.add_argument(
            "-bp",
            "--base_port",
            dest="base_port",
            type=int,
            default=self.base_port,
            help="set how base html port 80 to be exposed - incrementing by one for each version [default: %(default)s]",
        )
        parser.add_argument(
            "-vl",
            "--version_list",
            dest="versions",
            nargs="*",
            default=self.versions,
            help="mediawiki versions to create docker applications for [default: %(default)s] ",
        )

    def fromArgs(self, args):
        """
        initialize me from the given commmand line arguments

        Args:
            args(Namespace): the command line arguments
        """
        dbc_name = args.db_container_name
        if dbc_name:
            env = DockerMap.getEnv(dbc_name)
            self.mySQLRootPassword = env["MYSQL_ROOT_PASSWORD"]
            pass
        super().fromArgs(args)

addArgs(parser)

add my arguments to the given parser

Source code in mwdocker/config.py

def addArgs(self, parser):
    """
    add my arguments to the given parser
    """
    super().addArgs(parser)
    parser.add_argument(
        "-bp",
        "--base_port",
        dest="base_port",
        type=int,
        default=self.base_port,
        help="set how base html port 80 to be exposed - incrementing by one for each version [default: %(default)s]",
    )
    parser.add_argument(
        "-vl",
        "--version_list",
        dest="versions",
        nargs="*",
        default=self.versions,
        help="mediawiki versions to create docker applications for [default: %(default)s] ",
    )

fromArgs(args)

initialize me from the given commmand line arguments

Parameters:

Name	Type	Description	Default
args(Namespace)		the command line arguments	required

Source code in mwdocker/config.py

def fromArgs(self, args):
    """
    initialize me from the given commmand line arguments

    Args:
        args(Namespace): the command line arguments
    """
    dbc_name = args.db_container_name
    if dbc_name:
        env = DockerMap.getEnv(dbc_name)
        self.mySQLRootPassword = env["MYSQL_ROOT_PASSWORD"]
        pass
    super().fromArgs(args)

MwConfig

MediaWiki and docker configuration for a Single Wiki

Source code in mwdocker/config.py

@lod_storable
class MwConfig:
    """
    MediaWiki and docker configuration for a Single Wiki
    """

    version: str = "1.39.17"
    smw_version: Optional[str] = None
    extensionNameList: Optional[List[str]] = field(
        default_factory=lambda: [
            "Admin Links",
            "Header Tabs",
            "ParserFunctions",
            "SyntaxHighlight",
            "Variables",
        ]
    )
    extensionJsonFile: Optional[str] = None
    user: str = "Sysop"
    prefix: str = "mw"
    logo: str = "$wgResourceBasePath/resources/assets/wiki.png"
    url = None
    full_url = None
    prot: str = "http"
    host: str = Host.get_default_host()
    article_path: Optional[str] = None  # "/index.php/$1"
    script_path: str = ""
    wikiId: Optional[str] = None
    # mysql settings
    mySQLRootPassword: Optional[str] = None
    mySQLPassword: Optional[str] = None
    mariaDBVersion: str = "11.8"

    # docker settings
    bind_mount: bool = False
    port: int = 9080
    base_port: Optional[int] = None
    sql_port: int = 9306
    container_base_name: Optional[str] = None
    # derived from container_base_name if different than default
    # an external db_container is going to be used
    db_container_name: Optional[str] = None
    networkName: str = "mwNetwork"
    docker_path: Optional[str] = None
    gid: int = 33  # www-data
    uid: int = 33  # www-data

    # build control
    verbose: bool = True
    random_password: bool = False
    force_user: bool = False
    lenient: bool = True
    password_length: int = 15
    forceRebuild: bool = False
    debug: bool = False
    # FIXME - we should avoid a predefined known password
    password: str = "sysop-1234!"

    def default_docker_path(self) -> str:
        """
        get the default docker path

        Returns:
            str: $HOME/.pymediawikidocker
        """
        home = str(Path.home())
        docker_path = f"{home}/.pymediawikidocker"
        return docker_path

    def __post_init__(self):
        """
        post initialization configuration
        """
        self.fullVersion = f"MediaWiki {self.version}"
        self.underscoreVersion = self.version.replace(".", "_")
        self.shortVersion = self.getShortVersion()
        if not self.docker_path:
            self.docker_path = self.default_docker_path()
        if not self.container_base_name:
            self.container_base_name = f"{self.prefix}-{self.shortVersion}"
        if not self.db_container_name:
            self.db_container_name = f"{self.container_base_name}-db"
        if not self.article_path:
            self.article_path = ""
        if not self.base_port:
            self.base_port = self.port
        self.reset_url(self.url)

    @property
    def has_external_db(self) -> bool:
        """
        Check if using external database container

        Returns:
            bool: True if db_container_name differs from default pattern
        """
        default_db_name = f"{self.container_base_name}-db"
        external = self.db_container_name != default_db_name
        return external

    def reset_url(self, url: str):
        """
        reset my url

        Args:
            url(str): the url to set
        """
        if url:
            pr = urlparse(url)
            self.prot = pr.scheme
            self.host = pr.hostname
            self.script_path = pr.path
            self.base_url = f"{self.prot}://{self.host}"
            self.full_url = url
        else:
            self.base_url = f"{self.prot}://{self.host}:{self.base_port}"

            self.full_url = f"{self.base_url}{self.script_path}"

    def reset_container_base_name(self, container_base_name: str = None):
        """
        reset the container base name to the given name

        Args:
            container_base_name(str): the new container base name
        """
        self.container_base_name = container_base_name
        self.__post_init__()

    def as_dict(self) -> dict:
        """
        return my fields as a dict
        dataclasses to dict conversion convenienc and information hiding

        Returns:
            dict: my fields in dict format
        """
        config_dict = dataclasses.asdict(self)
        return config_dict

    def as_json(self) -> str:
        """
        return me as a json string

        Returns:
            str: my json representation
        """
        config_dict = self.as_dict()
        json_str = json.dumps(config_dict, indent=2)
        return json_str

    def get_config_path(self) -> str:
        """
        get my configuration base path

        Returns:
            str: the path to my configuration
        """
        config_base_path = f"{self.docker_path}/{self.container_base_name}"
        os.makedirs(config_base_path, exist_ok=True)
        path = f"{config_base_path}/MwConfig.json"
        return path

    def save(self, path: str = None) -> str:
        """
        save my json

        Args:
            path(str): the path to store to - if None use {docker_path}/{container_base_name}/MwConfig.json
        Returns:
            str: the path
        """
        if path is None:
            path = self.get_config_path()

        json_str = self.as_json()
        with open(path, "w") as f:
            print(json_str, file=f)
        return path

    def load(self, path: str = None) -> "MwConfig":
        """
        load the the MwConfig from the given path of if path is None (default)
        use the config_path for the current configuration

        restores the ExtensionMap on load

        Args:
            path(str): the path to load from

        Returns:
            MwConfig: a MediaWiki Configuration
        """
        if path is None:
            path = self.get_config_path()
        with open(path, "r") as json_file:
            json_str = json_file.read()
            config = self.__class__.from_json(json_str)
            # restore extension map
            config.getExtensionMap(config.extensionNameList, config.extensionJsonFile)
            return config

    def getShortVersion(self, separator=""):
        """
        get my short version e.g. convert 1.27.7 to 127

        Returns:
            str: the short version string
        """
        versionMatch = re.match(r"(?P<major>[0-9]+)\.(?P<minor>[0-9]+)", self.version)
        shortVersion = (
            f"{versionMatch.group('major')}{separator}{versionMatch.group('minor')}"
        )
        return shortVersion

    def create_random_password(self, length: int = 15) -> str:
        """
        create a random password

        Args:
            length(int): the length of the password

        Returns:
            str:a random password with the given length
        """
        random_password = secrets.token_urlsafe(length)
        return random_password

    def getWikiId(self):
        """
        get the wikiId

        Returns:
            str: e.g. mw-9080
        """
        if self.wikiId is None:
            wikiId = f"{self.prefix}-{self.port}"
        else:
            wikiId = self.wikiId
        return wikiId

    def getExtensionMap(
        self, extensionNameList: list = None, extensionJsonFile: str = None
    ):
        """
        get map of extensions to handle

        Args:
            extensionNameList (list): a list of extension names
            extensionJsonFile (str): the name of an extra extensionJsonFile (if any)
        """
        self.extensionMap = {}
        extensionList = ExtensionList.restore()
        self.extByName, duplicates = LOD.getLookup(extensionList.extensions, "name")
        if len(duplicates) > 0:
            print(f"{len(duplicates)} duplicate extensions: ")
            for duplicate in duplicates:
                print(duplicate.name)
        if extensionJsonFile is not None:
            extraExtensionList = ExtensionList.load_from_json_file(
                extensionJsonFile
            )  # @UndefinedVariable
            for ext in extraExtensionList.extensions:
                if ext.name in self.extByName:
                    print(f"overriding {ext.name} extension definition")
                self.extByName[ext.name] = ext
        if extensionNameList is not None:
            self.addExtensions(extensionNameList)
        return self.extensionMap

    def addExtensions(self, extensionNameList):
        """
        add extensions for the given list of extension names
        """
        for extensionName in extensionNameList:
            if extensionName in self.extByName:
                self.extensionMap[extensionName] = self.extByName[extensionName]
            else:
                print(f"warning: extension {extensionName} not known")

    def fromArgs(self, args):
        """
        initialize me from the given commmand line arguments

        Args:
            args(Namespace): the command line arguments
        """
        self.prefix = args.prefix
        self.article_path = args.article_path
        self.container_base_name = args.container_name
        self.db_container_name = args.db_container_name
        self.docker_path = args.docker_path
        self.extensionNameList = args.extensionNameList
        self.extensionJsonFile = args.extensionJsonFile
        self.bind_mount = args.bind_mount
        self.uid = args.uid
        self.gid = args.gid
        self.forceRebuild = args.forceRebuild or getattr(args, "force", False)
        self.host = args.host
        self.logo = args.logo
        self.mariaDBVersion = args.mariaDBVersion
        # passwords
        if args.mysqlRootPassword:
            self.mySQLRootPassword = args.mysqlRootPassword
        if not self.mySQLRootPassword:
            if args.db_container_name:
                # we need the password from the database container
                pass
            else:
                self.mySQLRootPassword = self.create_random_password(
                    self.password_length
                )
        if args.mysqlPassword:
            self.mySQLPassword = args.mysqlPassword
        else:
            self.mySQLPassword = self.create_random_password(self.password_length)
        self.prot = args.prot
        self.script_path = args.script_path
        self.versions = args.versions
        self.user = args.user
        self.random_password = args.random_password
        self.force_user = args.force_user
        self.lenient = args.lenient
        self.password = args.password
        self.password_length = args.password_length
        self.base_port = args.base_port
        self.sql_port = args.sql_port
        self.smw_version = args.smw_version
        self.verbose = not args.quiet
        self.debug = args.debug
        self.getExtensionMap(self.extensionNameList, self.extensionJsonFile)
        self.reset_url(args.url)

    def addArgs(self, parser):
        """
        add Arguments to the given parser
        """
        parser.add_argument(
            "--article_path",
            default=self.article_path,
            help="change to any article_path you might need to set [default: %(default)s]",
        )
        parser.add_argument(
            "-bm",
            "--bind-mount",
            action="store_true",
            help="use bind mounts instead of volumes",
        )
        parser.add_argument(
            "-cn",
            "--container_name",
            default=self.container_base_name,
            help="set container name (only valid and recommended for single version call)",
        )
        parser.add_argument(
            "-dcn",
            "--db_container_name",
            help="set database container name [default: %(default)s]",
        )
        parser.add_argument(
            "-el",
            "--extensionList",
            dest="extensionNameList",
            nargs="*",
            default=self.extensionNameList,
            help="list of extensions to be installed [default: %(default)s]",
        )
        parser.add_argument(
            "-ej",
            "--extensionJson",
            dest="extensionJsonFile",
            default=self.extensionJsonFile,
            help="additional extension descriptions default: [default: %(default)s]",
        )
        # since -f is a default options we have to make sure we accept it as forceRebuild
        parser.add_argument(
            "--forceRebuild",
            action="store_true",
            default=self.forceRebuild,
            help="force rebuilding  [default: %(default)s]",
        )
        parser.add_argument(
            "-fu",
            "--force_user",
            action="store_true",
            default=self.force_user,
            help="force overwrite of wikiuser",
        )
        parser.add_argument(
            "--host",
            default=Host.get_default_host(),
            help="the host to serve / listen from [default: %(default)s]",
        )
        parser.add_argument(
            "-dp",
            "--docker_path",
            default=self.default_docker_path(),
            help="the base directory to store docker and jinja template files [default: %(default)s]",
        )
        parser.add_argument(
            "--lenient",
            action="store_true",
            help="do not throw error on wikiuser difference",
        )
        parser.add_argument(
            "--logo", default=self.logo, help="set Logo [default: %(default)s]"
        )
        parser.add_argument(
            "-mv",
            "--mariaDBVersion",
            dest="mariaDBVersion",
            default=self.mariaDBVersion,
            help="mariaDB Version to be installed [default: %(default)s]",
        )
        parser.add_argument(
            "--mysqlRootPassword",
            default=self.mySQLRootPassword,
            help="set sql root Password [default: %(default)s] - random password if None",
        )
        parser.add_argument(
            "--mysqlPassword",
            default=self.mySQLPassword,
            help="set sql user Password [default: %(default)s] - random password if None",
        )
        parser.add_argument(
            "-rp",
            "--random_password",
            action="store_true",
            default=self.random_password,
            help="create random password and create wikiuser while at it",
        )
        parser.add_argument(
            "-p",
            "--password",
            dest="password",
            default=self.password,
            help="set password for initial user [default: %(default)s] ",
        )
        parser.add_argument(
            "-pl",
            "--password_length",
            default=self.password_length,
            help="set the password length for random passwords[default: %(default)s] ",
        )
        parser.add_argument(
            "--prefix",
            default=self.prefix,
            help="the container name prefix to use [default: %(default)s]",
        )
        parser.add_argument(
            "--prot",
            default=self.prot,
            help="change to https in case [default: %(default)s]",
        )
        parser.add_argument(
            "--script_path",
            default=self.script_path,
            help="change to any script_path you might need to set [default: %(default)s]",
        )
        parser.add_argument(
            "--url",
            default=self.url,
            help="will set prot host,script_path, and optionally port based on the url given [default: %(default)s]",
        )
        parser.add_argument(
            "-sp",
            "--sql_base_port",
            dest="sql_port",
            type=int,
            default=self.sql_port,
            help="set base mySql port 3306 to be exposed - incrementing by one for each version [default: %(default)s]",
        )
        parser.add_argument(
            "-smw",
            "--smw_version",
            dest="smw_version",
            default=self.smw_version,
            help="set SemanticMediaWiki Version to be installed default is None - no installation of SMW",
        )
        parser.add_argument(
            "-u",
            "--user",
            dest="user",
            default=self.user,
            help="set username of initial user with sysop rights [default: %(default)s] ",
        )
        parser.add_argument(
            "--uid",
            type=int,
            default=self.uid,
            help="User ID  (default: 33 for www-data)",
        )
        parser.add_argument(
            "--gid",
            type=int,
            default=self.gid,
            help="Group ID (default: 33 for www-data)",
        )

has_external_db property

Check if using external database container

Returns:

Name	Type	Description
bool	bool	True if db_container_name differs from default pattern

__post_init__()

post initialization configuration

Source code in mwdocker/config.py

def __post_init__(self):
    """
    post initialization configuration
    """
    self.fullVersion = f"MediaWiki {self.version}"
    self.underscoreVersion = self.version.replace(".", "_")
    self.shortVersion = self.getShortVersion()
    if not self.docker_path:
        self.docker_path = self.default_docker_path()
    if not self.container_base_name:
        self.container_base_name = f"{self.prefix}-{self.shortVersion}"
    if not self.db_container_name:
        self.db_container_name = f"{self.container_base_name}-db"
    if not self.article_path:
        self.article_path = ""
    if not self.base_port:
        self.base_port = self.port
    self.reset_url(self.url)

addArgs(parser)

add Arguments to the given parser

Source code in mwdocker/config.py

def addArgs(self, parser):
    """
    add Arguments to the given parser
    """
    parser.add_argument(
        "--article_path",
        default=self.article_path,
        help="change to any article_path you might need to set [default: %(default)s]",
    )
    parser.add_argument(
        "-bm",
        "--bind-mount",
        action="store_true",
        help="use bind mounts instead of volumes",
    )
    parser.add_argument(
        "-cn",
        "--container_name",
        default=self.container_base_name,
        help="set container name (only valid and recommended for single version call)",
    )
    parser.add_argument(
        "-dcn",
        "--db_container_name",
        help="set database container name [default: %(default)s]",
    )
    parser.add_argument(
        "-el",
        "--extensionList",
        dest="extensionNameList",
        nargs="*",
        default=self.extensionNameList,
        help="list of extensions to be installed [default: %(default)s]",
    )
    parser.add_argument(
        "-ej",
        "--extensionJson",
        dest="extensionJsonFile",
        default=self.extensionJsonFile,
        help="additional extension descriptions default: [default: %(default)s]",
    )
    # since -f is a default options we have to make sure we accept it as forceRebuild
    parser.add_argument(
        "--forceRebuild",
        action="store_true",
        default=self.forceRebuild,
        help="force rebuilding  [default: %(default)s]",
    )
    parser.add_argument(
        "-fu",
        "--force_user",
        action="store_true",
        default=self.force_user,
        help="force overwrite of wikiuser",
    )
    parser.add_argument(
        "--host",
        default=Host.get_default_host(),
        help="the host to serve / listen from [default: %(default)s]",
    )
    parser.add_argument(
        "-dp",
        "--docker_path",
        default=self.default_docker_path(),
        help="the base directory to store docker and jinja template files [default: %(default)s]",
    )
    parser.add_argument(
        "--lenient",
        action="store_true",
        help="do not throw error on wikiuser difference",
    )
    parser.add_argument(
        "--logo", default=self.logo, help="set Logo [default: %(default)s]"
    )
    parser.add_argument(
        "-mv",
        "--mariaDBVersion",
        dest="mariaDBVersion",
        default=self.mariaDBVersion,
        help="mariaDB Version to be installed [default: %(default)s]",
    )
    parser.add_argument(
        "--mysqlRootPassword",
        default=self.mySQLRootPassword,
        help="set sql root Password [default: %(default)s] - random password if None",
    )
    parser.add_argument(
        "--mysqlPassword",
        default=self.mySQLPassword,
        help="set sql user Password [default: %(default)s] - random password if None",
    )
    parser.add_argument(
        "-rp",
        "--random_password",
        action="store_true",
        default=self.random_password,
        help="create random password and create wikiuser while at it",
    )
    parser.add_argument(
        "-p",
        "--password",
        dest="password",
        default=self.password,
        help="set password for initial user [default: %(default)s] ",
    )
    parser.add_argument(
        "-pl",
        "--password_length",
        default=self.password_length,
        help="set the password length for random passwords[default: %(default)s] ",
    )
    parser.add_argument(
        "--prefix",
        default=self.prefix,
        help="the container name prefix to use [default: %(default)s]",
    )
    parser.add_argument(
        "--prot",
        default=self.prot,
        help="change to https in case [default: %(default)s]",
    )
    parser.add_argument(
        "--script_path",
        default=self.script_path,
        help="change to any script_path you might need to set [default: %(default)s]",
    )
    parser.add_argument(
        "--url",
        default=self.url,
        help="will set prot host,script_path, and optionally port based on the url given [default: %(default)s]",
    )
    parser.add_argument(
        "-sp",
        "--sql_base_port",
        dest="sql_port",
        type=int,
        default=self.sql_port,
        help="set base mySql port 3306 to be exposed - incrementing by one for each version [default: %(default)s]",
    )
    parser.add_argument(
        "-smw",
        "--smw_version",
        dest="smw_version",
        default=self.smw_version,
        help="set SemanticMediaWiki Version to be installed default is None - no installation of SMW",
    )
    parser.add_argument(
        "-u",
        "--user",
        dest="user",
        default=self.user,
        help="set username of initial user with sysop rights [default: %(default)s] ",
    )
    parser.add_argument(
        "--uid",
        type=int,
        default=self.uid,
        help="User ID  (default: 33 for www-data)",
    )
    parser.add_argument(
        "--gid",
        type=int,
        default=self.gid,
        help="Group ID (default: 33 for www-data)",
    )

addExtensions(extensionNameList)

add extensions for the given list of extension names

Source code in mwdocker/config.py

def addExtensions(self, extensionNameList):
    """
    add extensions for the given list of extension names
    """
    for extensionName in extensionNameList:
        if extensionName in self.extByName:
            self.extensionMap[extensionName] = self.extByName[extensionName]
        else:
            print(f"warning: extension {extensionName} not known")

as_dict()

return my fields as a dict dataclasses to dict conversion convenienc and information hiding

Returns:

Name	Type	Description
dict	dict	my fields in dict format

Source code in mwdocker/config.py

def as_dict(self) -> dict:
    """
    return my fields as a dict
    dataclasses to dict conversion convenienc and information hiding

    Returns:
        dict: my fields in dict format
    """
    config_dict = dataclasses.asdict(self)
    return config_dict

as_json()

return me as a json string

Returns:

Name	Type	Description
str	str	my json representation

Source code in mwdocker/config.py

def as_json(self) -> str:
    """
    return me as a json string

    Returns:
        str: my json representation
    """
    config_dict = self.as_dict()
    json_str = json.dumps(config_dict, indent=2)
    return json_str

create_random_password(length=15)

create a random password

Parameters:

Name	Type	Description	Default
length(int)		the length of the password	required

Returns:

Name	Type	Description
str	str	a random password with the given length

Source code in mwdocker/config.py

def create_random_password(self, length: int = 15) -> str:
    """
    create a random password

    Args:
        length(int): the length of the password

    Returns:
        str:a random password with the given length
    """
    random_password = secrets.token_urlsafe(length)
    return random_password

default_docker_path()

get the default docker path

Returns:

Name	Type	Description
str	str	$HOME/.pymediawikidocker

Source code in mwdocker/config.py

def default_docker_path(self) -> str:
    """
    get the default docker path

    Returns:
        str: $HOME/.pymediawikidocker
    """
    home = str(Path.home())
    docker_path = f"{home}/.pymediawikidocker"
    return docker_path

fromArgs(args)

initialize me from the given commmand line arguments

Parameters:

Name	Type	Description	Default
args(Namespace)		the command line arguments	required

Source code in mwdocker/config.py

def fromArgs(self, args):
    """
    initialize me from the given commmand line arguments

    Args:
        args(Namespace): the command line arguments
    """
    self.prefix = args.prefix
    self.article_path = args.article_path
    self.container_base_name = args.container_name
    self.db_container_name = args.db_container_name
    self.docker_path = args.docker_path
    self.extensionNameList = args.extensionNameList
    self.extensionJsonFile = args.extensionJsonFile
    self.bind_mount = args.bind_mount
    self.uid = args.uid
    self.gid = args.gid
    self.forceRebuild = args.forceRebuild or getattr(args, "force", False)
    self.host = args.host
    self.logo = args.logo
    self.mariaDBVersion = args.mariaDBVersion
    # passwords
    if args.mysqlRootPassword:
        self.mySQLRootPassword = args.mysqlRootPassword
    if not self.mySQLRootPassword:
        if args.db_container_name:
            # we need the password from the database container
            pass
        else:
            self.mySQLRootPassword = self.create_random_password(
                self.password_length
            )
    if args.mysqlPassword:
        self.mySQLPassword = args.mysqlPassword
    else:
        self.mySQLPassword = self.create_random_password(self.password_length)
    self.prot = args.prot
    self.script_path = args.script_path
    self.versions = args.versions
    self.user = args.user
    self.random_password = args.random_password
    self.force_user = args.force_user
    self.lenient = args.lenient
    self.password = args.password
    self.password_length = args.password_length
    self.base_port = args.base_port
    self.sql_port = args.sql_port
    self.smw_version = args.smw_version
    self.verbose = not args.quiet
    self.debug = args.debug
    self.getExtensionMap(self.extensionNameList, self.extensionJsonFile)
    self.reset_url(args.url)

getExtensionMap(extensionNameList=None, extensionJsonFile=None)

get map of extensions to handle

Parameters:

Name	Type	Description	Default
extensionNameList	list	a list of extension names	None
extensionJsonFile	str	the name of an extra extensionJsonFile (if any)	None

Source code in mwdocker/config.py

def getExtensionMap(
    self, extensionNameList: list = None, extensionJsonFile: str = None
):
    """
    get map of extensions to handle

    Args:
        extensionNameList (list): a list of extension names
        extensionJsonFile (str): the name of an extra extensionJsonFile (if any)
    """
    self.extensionMap = {}
    extensionList = ExtensionList.restore()
    self.extByName, duplicates = LOD.getLookup(extensionList.extensions, "name")
    if len(duplicates) > 0:
        print(f"{len(duplicates)} duplicate extensions: ")
        for duplicate in duplicates:
            print(duplicate.name)
    if extensionJsonFile is not None:
        extraExtensionList = ExtensionList.load_from_json_file(
            extensionJsonFile
        )  # @UndefinedVariable
        for ext in extraExtensionList.extensions:
            if ext.name in self.extByName:
                print(f"overriding {ext.name} extension definition")
            self.extByName[ext.name] = ext
    if extensionNameList is not None:
        self.addExtensions(extensionNameList)
    return self.extensionMap

getShortVersion(separator='')

get my short version e.g. convert 1.27.7 to 127

Returns:

Name	Type	Description
str		the short version string

Source code in mwdocker/config.py

def getShortVersion(self, separator=""):
    """
    get my short version e.g. convert 1.27.7 to 127

    Returns:
        str: the short version string
    """
    versionMatch = re.match(r"(?P<major>[0-9]+)\.(?P<minor>[0-9]+)", self.version)
    shortVersion = (
        f"{versionMatch.group('major')}{separator}{versionMatch.group('minor')}"
    )
    return shortVersion

getWikiId()

get the wikiId

Returns:

Name	Type	Description
str		e.g. mw-9080

Source code in mwdocker/config.py

def getWikiId(self):
    """
    get the wikiId

    Returns:
        str: e.g. mw-9080
    """
    if self.wikiId is None:
        wikiId = f"{self.prefix}-{self.port}"
    else:
        wikiId = self.wikiId
    return wikiId

get_config_path()

get my configuration base path

Returns:

Name	Type	Description
str	str	the path to my configuration

Source code in mwdocker/config.py

def get_config_path(self) -> str:
    """
    get my configuration base path

    Returns:
        str: the path to my configuration
    """
    config_base_path = f"{self.docker_path}/{self.container_base_name}"
    os.makedirs(config_base_path, exist_ok=True)
    path = f"{config_base_path}/MwConfig.json"
    return path

load(path=None)

load the the MwConfig from the given path of if path is None (default) use the config_path for the current configuration

restores the ExtensionMap on load

Parameters:

Name	Type	Description	Default
path(str)		the path to load from	required

Returns:

Name	Type	Description
MwConfig	MwConfig	a MediaWiki Configuration

Source code in mwdocker/config.py

def load(self, path: str = None) -> "MwConfig":
    """
    load the the MwConfig from the given path of if path is None (default)
    use the config_path for the current configuration

    restores the ExtensionMap on load

    Args:
        path(str): the path to load from

    Returns:
        MwConfig: a MediaWiki Configuration
    """
    if path is None:
        path = self.get_config_path()
    with open(path, "r") as json_file:
        json_str = json_file.read()
        config = self.__class__.from_json(json_str)
        # restore extension map
        config.getExtensionMap(config.extensionNameList, config.extensionJsonFile)
        return config

reset_container_base_name(container_base_name=None)

reset the container base name to the given name

Parameters:

Name	Type	Description	Default
container_base_name(str)		the new container base name	required

Source code in mwdocker/config.py

def reset_container_base_name(self, container_base_name: str = None):
    """
    reset the container base name to the given name

    Args:
        container_base_name(str): the new container base name
    """
    self.container_base_name = container_base_name
    self.__post_init__()

reset_url(url)

reset my url

Parameters:

Name	Type	Description	Default
url(str)		the url to set	required

Source code in mwdocker/config.py

def reset_url(self, url: str):
    """
    reset my url

    Args:
        url(str): the url to set
    """
    if url:
        pr = urlparse(url)
        self.prot = pr.scheme
        self.host = pr.hostname
        self.script_path = pr.path
        self.base_url = f"{self.prot}://{self.host}"
        self.full_url = url
    else:
        self.base_url = f"{self.prot}://{self.host}:{self.base_port}"

        self.full_url = f"{self.base_url}{self.script_path}"

save(path=None)

save my json

Parameters:

Name	Type	Description	Default
path(str)		the path to store to - if None use {docker_path}/{container_base_name}/MwConfig.json	required

Returns: str: the path

Source code in mwdocker/config.py

def save(self, path: str = None) -> str:
    """
    save my json

    Args:
        path(str): the path to store to - if None use {docker_path}/{container_base_name}/MwConfig.json
    Returns:
        str: the path
    """
    if path is None:
        path = self.get_config_path()

    json_str = self.as_json()
    with open(path, "w") as f:
        print(json_str, file=f)
    return path

docker

Created on 2021-08-06

@author: wf

DBStatus dataclass

the Database Status

Source code in mwdocker/docker.py

@dataclass
class DBStatus:
    """
    the Database Status
    """

    attempts: int
    max_tries: int
    ok: bool
    msg: str
    ex: typing.Optional[Exception] = None

DockerApplication

Bases: object

MediaWiki Docker image

Source code in mwdocker/docker.py

class DockerApplication(object):
    """
    MediaWiki Docker image
    """

    def __init__(self, config: MwClusterConfig):
        """
        Constructor

        Args:
            config: MwClusterConfig,
            home: the home directory to use
        """
        self.config = config
        # branch as need for git clone e.g. https://gerrit.wikimedia.org/g/mediawiki/extensions/MagicNoCache
        self.branch = f"REL{self.config.getShortVersion('_')}"
        self.composerVersion = 1
        if self.config.shortVersion >= "139":
            self.composerVersion = 2
        # jinja and docker prerequisites
        self.env = self.getJinjaEnv()
        # docker file location
        self.docker_path = (
            f"{self.config.docker_path}/{self.config.container_base_name}"
        )
        os.makedirs(self.docker_path, exist_ok=True)

        self.getContainers()
        self.dbConn = None
        self.wiki_id = self.config.getWikiId()
        self.database = f"{self.wiki_id}_wiki"
        self.dbUser = f"{self.wiki_id}_user"
        self.wikiUser = None
        # Hook to allow modifying results
        # e.g. docker-compose.yaml after generation is finished
        self.postgen_hook = None

    @staticmethod
    def checkDockerEnvironment(debug: bool = False) -> str:
        """
        check the docker environment

        Args:
            debug (bool): if True show debug information

        Returns:
            str: an error message or None
        """
        errMsg = None
        os_path = os.environ["PATH"]
        paths = ["/usr/local/bin"]
        for path in paths:
            if os.path.islink(f"{path}/docker"):
                if not path in os_path:
                    os.environ["PATH"] = f"{os_path}{os.pathsep}{path}"
                    if debug:
                        print(
                            f"""modified PATH from {os_path} to \n{os.environ["PATH"]}"""
                        )
        if not docker.compose.is_installed():
            errMsg = """docker compose up needs to be working"""

        return errMsg

    def get_version_url(self, host_port: str) -> str:
        """
        get the Special:Version url
        """
        url = self.config.full_url
        # fix url to local port
        # @TODO isn't this superfluous / has no effect ...?
        url = url.replace(str(self.config.port), host_port)
        version_url = f"{url}/index.php?title=Special:Version"
        return version_url

    def check(self) -> int:
        """
        check me

        Returns:
            int: exitCode: 0 if ok, 1 if not ok
        """
        DockerApplication.checkDockerEnvironment(self.config.debug)
        exitCode = 0
        mw, db = self.getContainers()
        if not mw:
            print("mediawiki container missing")
            exitCode = 1
        if not db:
            print("database container missing")
            exitCode = 1
        if mw and db and mw.check() and db.check():
            host_port = mw.getHostPort(80)
            if host_port:
                Logger.check_and_log_equal(
                    f"port binding", host_port, "expected  port", str(self.config.port)
                )
                version_url = self.get_version_url(host_port)
                ok = self.checkWiki(version_url)
                if not ok:
                    exitCode = 1
            else:
                self.log(f"port binding for port 80 missing", False)
                exitCode = 1
            pass
        return exitCode

    def checkWiki(self, version_url: str) -> bool:
        """
        check this wiki against the content of the given version_url
        """
        print(f"Checking {version_url} ...")
        ok = True
        try:
            html_tables = HtmlTables(version_url)
            tables = html_tables.get_tables("h2")
            if self.config.debug:
                p = pprint.PrettyPrinter(indent=2)
                p.pprint(tables)
            ok = ok and Logger.check_and_log(
                "Special Version accessible ...", "Installed software" in tables
            )
            if ok:
                software = tables["Installed software"]
                software_map, _dup = LOD.getLookup(
                    software, "Product", withDuplicates=False
                )
                mw_version = software_map["MediaWiki"]["Version"]
                ok = ok and Logger.check_and_log_equal(
                    "Mediawiki Version", mw_version, "expected ", self.config.version
                )
                db_version_str = software_map["MariaDB"]["Version"]
                db_version = MariaDB.getVersion(db_version_str)
                ok = ok and Logger.check_and_log(
                    f"Maria DB Version {db_version} fitting expected {self.config.mariaDBVersion}?",
                    self.config.mariaDBVersion.startswith(db_version),
                )
                pass
        except Exception as ex:
            ok = Logger.check_and_log(str(ex), False)
        return ok

    def getContainerName(self, kind: str, separator: str):
        """
        get my container Name
        """
        containerName = f"{self.config.container_base_name}{separator}{kind}"
        return containerName

    def getContainers(self):
        """
        get my containers

        Returns:
            Tuple(
        """
        self.dbContainer = None
        self.mwContainer = None
        containerMap = DockerMap.getContainerMap()
        for separator in ["-", "_"]:
            dbContainerName = self.getContainerName("db", separator)
            mwContainerName = self.getContainerName("mw", separator)
            if dbContainerName in containerMap:
                self.dbContainer = DockerContainer(
                    dbContainerName, "database", containerMap[dbContainerName]
                )
            if mwContainerName in containerMap:
                self.mwContainer = DockerContainer(
                    mwContainerName, "webserver", containerMap[mwContainerName]
                )
        return self.mwContainer, self.dbContainer

    def getJinjaEnv(self):
        """
        get a Jinja2 environment
        """
        scriptdir = os.path.dirname(os.path.realpath(__file__))
        resourcePath = os.path.realpath(f"{scriptdir}/resources")
        template_dir = os.path.realpath(f"{resourcePath}/templates")
        # print(f"jinja template directory is {template_dir}")
        env = Environment(loader=FileSystemLoader(template_dir))
        return env

    def createWikiUser(self, wikiId: str = None, store: bool = False):
        """
        create my wikiUser and optionally save it

        Args:
           store (bool): if True save my user data to the relevant ini File
        """
        if not wikiId:
            wikiId = f"{self.config.container_base_name}"
        userDict = {
            "wikiId": f"{wikiId}",
            "url": f"{self.config.base_url}:{self.config.port}",
            "scriptPath": f"{self.config.script_path}",
            "user": f"{self.config.user}",
            "email": "noreply@nouser.com",
            "version": f"{self.config.fullVersion}",
            "password": f"{self.config.password}",
        }
        wikiUser = WikiUser.ofDict(userDict, encrypted=False)
        if store:
            wikiUser.save()
        return wikiUser

    def createOrModifyWikiUser(
        self, wikiId, force_overwrite: bool = False, lenient: bool = False
    ) -> WikiUser:
        """
        create or modify the WikiUser for this DockerApplication

        Args:
            wikiId (str): the wikiId to create or modify a wiki user for
            force_overwrite (bool): if True overwrite the wikiuser info
            lenient(bool): do not throw Exception if wikiuser exists
        """
        wikiUsers = WikiUser.getWikiUsers(lenient=True)
        if wikiId in wikiUsers and not force_overwrite:
            wikiUser = wikiUsers[wikiId]
            if self.config.password != wikiUser.getPassword():
                msg = f"wikiUser for wiki {wikiId} already exists but with different password"
                if lenient:
                    print(msg, file=sys.stderr)
                else:
                    raise Exception(msg)
            pass
        else:
            wikiUser = self.createWikiUser(wikiId, store=True)
        return wikiUser

    def execute(self, *commands: str):
        """
        Execute the given variable list of command strings inside the MediaWiki container.

        Args:
            commands: str - command parts to be executed
        """
        if not self.mwContainer:
            mwContainerNameDash = self.getContainerName("mw", "-")
            mwContainerNameUnderscore = self.getContainerName("mw", "_")
            raise Exception(
                f"no mediawiki Container {mwContainerNameDash} or {mwContainerNameUnderscore} for {self.name} "
                f"activated by docker compose\n- you might want to check the separator character used "
                f"for container names for your platform {platform.system()}"
            )
        self.mwContainer.execute(*commands, verbose=self.config.verbose)

    def close(self):
        """
        close the database
        """
        self.dbClose()

    def sqlQuery(self, query):
        """
        run the given SQL query
        """
        if self.dbConn and self.dbConn.is_connected():
            cursor = self.dbConn.cursor()
            cursor.execute(query)
            rows = cursor.fetchall()
            cursor.close()
            return rows
        else:
            if self.config.verbose:
                print(
                    f"Connection to {self.database} on {self.config.host} with user {self.dbUser} not established"
                )
            return None

    def dbClose(self):
        """
        close the database connection
        """
        if self.dbConn and self.dbConn.is_connected():
            self.dbConn.close()

    def dbConnect(self, timeout: int = 10):
        """
        connect to the database and return the connection

        Args:
            timeout (int): number of seconds for timeout

        Returns:
            the connection
        """
        if self.dbConn is None:
            try:
                self.dbConn = mysql.connector.connect(
                    host=self.config.host,
                    database=self.database,
                    user=self.dbUser,
                    port=self.config.sql_port,
                    password=self.config.mySQLPassword,
                    connection_timeout=timeout,
                )

            except Error as e:
                errMsg = str(e)
                print(
                    f"Connection to {self.database} on {self.config.host} with user {self.dbUser} failed error: {errMsg}"
                )
                if "Access denied" in errMsg:
                    raise e
        return self.dbConn

    def doCheckDBConnection(self, dbStatus: DBStatus, timeout: int = 10):
        """
        check the database connection of this application

        Args:
            timeout (int): how many seconds to wait

        Returns:
            DBStatus
        """
        dbStatus.attempts += 1
        self.dbConnect(timeout=timeout)
        if self.dbConn and self.dbConn.is_connected():
            rows = self.sqlQuery("select database();")
            dbStatus.ok = True
            if self.config.verbose:
                print(f"{dbStatus.msg} established database returns: {rows[0]}")

    def checkDBConnection(
        self,
        timeout: float = 10,
        initialSleep: float = 4.0,
        factor=1.5,
        maxTries: int = 9,
    ) -> DBStatus:
        """
        check database connection with retries

        Args:
            timeout (float): number of seconds for timeout
            initialSleep (float): number of seconds to initially wait/sleep
            maxTries (int): maximum number of retries before giving up between each try a sleep is done that starts
            with 0.5 secs and multiplies on every retry

        Returns:
            dbStatus: the status
        """
        conn_msg = f"SQL-Connection to {self.database} on {self.config.host} port {self.config.sql_port} with user {self.dbUser}"
        dbStatus = DBStatus(attempts=0, ok=False, msg=conn_msg, max_tries=maxTries)
        if self.config.verbose:
            print(
                f"Trying {dbStatus.msg} with max {maxTries} tries and {timeout}s timeout per try - initial sleep {initialSleep}s"
            )
        time.sleep(initialSleep)
        sleep = 2.0
        while not dbStatus.ok and dbStatus.attempts <= maxTries:
            try:
                self.doCheckDBConnection(dbStatus, timeout=timeout)
                if not dbStatus.ok:
                    if self.config.verbose:
                        print(
                            f"Connection attempt #{dbStatus.attempts}/{dbStatus.max_tries} failed will retry in {sleep:4.1f} secs"
                        )
                    # wait before trying
                    time.sleep(sleep)
                    sleep = sleep * factor
            except Exception as ex:
                dbStatus.ex = ex
                if self.config.verbose:
                    print(
                        f"Connection attempt #{dbStatus.attempts} failed with exception {str(ex)} - will not retry ..."
                    )
                if self.config.debug:
                    print(traceback.format_exc())
                break
        return dbStatus

    def optionalWrite(self, targetPath: str, content: str, overwrite: bool = False):
        """
        optionally Write the modified content to the given targetPath

        Args:
            targetPath (str): the path to write the content to
            content (str): the content to write
            overwrite (bool): if True overwrite the existing content
        """
        if not overwrite and os.path.isfile(targetPath):
            if self.config.verbose:
                print(f"{targetPath} already exists!")
            return
        with open(targetPath, "w", newline="") as targetFile:
            targetFile.write(content)

    def generate(
        self, templateName: str, targetPath: str, overwrite: bool = False, **kwArgs
    ):
        """
        generate file at targetPath using the given templateName

        Args:
            templateName (str): the Jinja2 template to use
            targetPath (str): the path to the target file
            overwrite (bool): if True overwrite existing files
            kwArgs(): generic keyword arguments to pass on to template rendering
        """
        try:
            template = self.env.get_template(templateName)
            timestamp = datetime.datetime.now().isoformat()
            content = template.render(
                mwVersion=self.config.version,
                mariaDBVersion=self.config.mariaDBVersion,
                port=self.config.port,
                sql_port=self.config.sql_port,
                smw_version=self.config.smw_version,
                timestamp=timestamp,
                pmwdVersion=Version.version,
                config=self.config,
                **kwArgs,
            )
            self.optionalWrite(targetPath, content, overwrite)

        except TemplateNotFound:
            print(f"no template {templateName} for {self.config.version}")

    def getComposerRequire(self):
        """
        get the json string for the composer require e.g. composer.local.json
        """
        requires = []
        for ext in self.config.extensionMap.values():
            # get the composer statement
            if ext.composer:
                requires.append(ext.composer)
        indent = "     "
        delim = "" if len(requires) == 0 else ",\n"
        requireList = ""
        if self.config.smw_version:
            requireList += f'{indent}"mediawiki/semantic-media-wiki": "~{self.config.smw_version}"{delim}'
        for i, require in enumerate(requires):
            delim = "" if i >= len(requires) - 1 else ",\n"
            requireList += f"{indent}{require}{delim}"
        requireJson = f"""{{
  "require": {{
{requireList}
  }}
}}"""
        return requireJson

    def genComposerRequire(self, composerFilePath, overwrite: bool = False):
        """
        gen the composer.local.json require file

        Args:
            composerFilePath (str): the name of the file to generate
        """
        requireJson = self.getComposerRequire()
        self.optionalWrite(composerFilePath, requireJson, overwrite)

    def generateAll(self, overwrite: bool = False):
        """
        generate all files needed for the docker handling

        Args:
            overwrite (bool): if True overwrite the existing files
        """
        # make sure we have the wiki_id ready
        wiki_id = self.config.getWikiId()
        # we have to configure whether
        # bind mounts or volumes are to be used
        if self.config.bind_mount:
            volume_type = "bind"
            mysql_data = f"/var/lib/mediawiki/mysql/{self.config.shortVersion}"
            wiki_sites = f"/var/www/mediawiki/sites"
        else:
            volume_type = "volume"
            mysql_data = "mysql-data"
            wiki_sites = "wiki-sites"
        # first generate Dockerfile
        # the goal is to get an empty MediaWiki (no LocalSettings/extensions)
        # with composer ready
        self.generate(
            "mwDockerfile",
            f"{self.docker_path}/Dockerfile",
            composerVersion=self.composerVersion,
            volume_type=volume_type,
            overwrite=overwrite,
        )
        # the master setup script
        # this used to be part of Dockerfile but
        # needs to be scripted when we use bind mounts due
        # to docker's poor design of permission and mount handling
        self.generate(
            f"setup-mediawiki.sh",
            f"{self.docker_path}/setup-mediawiki.sh",
            script_dir="/scripts",
            web_dir="/var/www/html",
            overwrite=overwrite,
        )
        # the Docker compose
        # at this stage we will have two containers
        # one for the database and one for the MediaWiki
        # the db container may be optionally an existing database container
        template_name = (
            "mwComposeExternalDB.yml"
            if self.config.has_external_db
            else "mwCompose.yml"
        )
        self.generate(
            template_name,
            f"{self.docker_path}/docker-compose.yml",
            # might be None for ExternalDB case
            mySQLRootPassword=self.config.mySQLRootPassword,
            mySQLPassword=self.config.mySQLPassword,
            container_base_name=self.config.container_base_name,
            db_container_name=self.config.db_container_name,
            wiki_id=wiki_id,
            volume_type=volume_type,
            mysql_data=mysql_data,
            wiki_sites=wiki_sites,
            scripts_dir=self.docker_path,
            uid=self.config.uid,
            gid=self.config.gid,
            overwrite=overwrite,
        )
        # now generate the parts we will use later to
        # create the fully configured wiki
        # first - LocalSettings with references to all extensions
        # secretKey
        # https://www.mediawiki.org/wiki/Manual:$wgSecretKey
        # 64 char random hex
        secretKey = secrets.token_hex(32)
        # upgradeKey
        # https://www.mediawiki.org/wiki/Manual:$wgUpgradeKey
        # 16 char random
        upgradeKey = secrets.token_hex(8)
        self.generate(
            f"mwLocalSettings{self.config.shortVersion}.php",
            f"{self.docker_path}/LocalSettings.php",
            wiki_id=wiki_id,
            mySQLPassword=self.config.mySQLPassword,
            hostname=self.config.host,
            extensions=self.config.extensionMap.values(),
            mwShortVersion=self.config.shortVersion,
            logo=self.config.logo,
            secretKey=secretKey,
            upgradeKey=upgradeKey,
            overwrite=overwrite,
        )
        # the SQL file for initial content
        self.generate(
            f"mwWiki{self.config.shortVersion}.sql",
            f"{self.docker_path}/wiki.sql",
            overwrite=overwrite,
        )
        # a WikiUser for automated access via
        # py-3rdparty mediawiki
        if self.config.random_password:
            self.config.password = self.config.create_random_password(
                length=self.config.password_length
            )
            if self.config.wikiId:
                self.createOrModifyWikiUser(
                    self.config.wikiId,
                    force_overwrite=self.config.force_user,
                    lenient=self.config.lenient,
                )
        self.generate(
            f"addSysopUser.sh",
            f"{self.docker_path}/addSysopUser.sh",
            user=self.config.user,
            password=self.config.password,
            overwrite=overwrite,
        )
        self.generate(
            f"installExtensions.sh",
            f"{self.docker_path}/installExtensions.sh",
            extensions=self.config.extensionMap.values(),
            branch=self.branch,
            overwrite=overwrite,
        )
        self.genComposerRequire(
            f"{self.docker_path}/composer.local.json", overwrite=overwrite
        )
        for file_name in [
            "phpinfo.php",
            "disable_sudo.sh",
            "install_djvu.sh",
            "plantuml.sh",
            "upload.ini",
        ]:
            self.generate(
                f"{file_name}", f"{self.docker_path}/{file_name}", overwrite=overwrite
            )

        # chmod generated scripts that need to be bash callable
        # to be executable on container operating system (linux)
        # when bind mounted or being copied
        for executable in [
            "disable_sudo.sh",
            "install_djvu.sh",
            "plantuml.sh",
            "addSysopUser.sh",
            "installExtensions.sh",
            "setup-mediawiki.sh",
        ]:
            path = os.path.join(self.docker_path, executable)
            if os.path.exists(path):
                st = os.stat(path)
                os.chmod(path, st.st_mode | stat.S_IXUSR | stat.S_IXGRP | stat.S_IXOTH)

        if self.postgen_hook:
            self.postgen_hook(self)

        # remember the configuration we used for generating
        # avoid endless loop - forceRebuilds - we have rebuild already
        forceRebuild = self.config.forceRebuild
        self.config.forceRebuild = False
        self.config.save()
        self.config.forceRebuild = forceRebuild

    def down(self, forceRebuild: bool = False):
        """
        run docker compose down

        see https://docs.docker.com/engine/reference/commandline/compose_down/
        and https://gabrieldemarmiesse.github.io/python-on-whales/sub-commands/compose/#down

        """
        DockerApplication.checkDockerEnvironment(self.config.debug)
        # change directory so that docker CLI will find the relevant dockerfile and docker-compose.yml
        if self.config.verbose:
            print(
                f"running docker compose down for {self.config.container_base_name} {self.config.version} docker application ..."
            )
        # remember current directory
        cwd = os.getcwd()
        os.chdir(self.docker_path)
        try:
            docker.compose.down(volumes=forceRebuild)
        except DockerException as dex:
            print(
                f"warning: docker compose down failed in {self.docker_path}:{str(dex)}"
            )
            pass
        # switch back to previous current directory
        os.chdir(cwd)

    def up(self, forceRebuild: bool = False):
        """
        start this docker application

        Args:
            forceRebuild (bool): if true stop and remove the existing containers
        """
        DockerApplication.checkDockerEnvironment(self.config.debug)
        if self.config.verbose:
            print(
                f"starting {self.config.container_base_name} {self.config.version} docker application ..."
            )
        if forceRebuild:
            for docker_container in [self.dbContainer, self.mwContainer]:
                if docker_container is not None:
                    container = docker_container.container
                    try:
                        container_name = container.name
                        if self.config.verbose:
                            print(f"stopping and removing container {container_name}")
                    except Exception as container_ex:
                        container = None
                    if container:
                        try:
                            container.stop()
                        except Exception as stop_ex:
                            if self.config.verbose:
                                print(f"stop failed with {str(stop_ex)}")
                            pass
                        try:
                            container.remove()
                        except Exception as remove_ex:
                            if self.config.verbose:
                                print(f"removed failed with {str(remove_ex)}")
                            pass
                    pass

        # remember current directory
        cwd = os.getcwd()

        # change directory so that docker CLI will find the relevant dockerfile and docker-compose.yml
        os.chdir(self.docker_path)
        # project_config = docker.compose.config()
        if forceRebuild:
            docker.compose.build()
        # run docker compose up
        # this might take a while e.g. downloading
        # run docker compose up
        try:
            docker.compose.up(detach=True, force_recreate=forceRebuild)
        except Exception as de:
            print(f"docker compose up failed in {self.docker_path}")
            raise de
            pass
        # switch back to previous current directory
        os.chdir(cwd)

        # check the startup of both containers
        mw, db = self.getContainers()
        for dc in [mw, db]:
            if dc:
                start_secs = dc.wait_for_state(running=True)
                if self.config.verbose:
                    print(f"{dc.name} 🟢 started in {start_secs:.2f}s")
        return mw, db

    def prepare_external_db_access(
        self, network_name: str = "db", db_alias: str = "db"
    ):
        """
        make sure the external database container can be reached
        """
        # ensure network exists
        nets = {n.name for n in docker.network.list()}
        if network_name not in nets:
            docker.network.create(network_name)

        # connect external DB container with alias
        try:
            docker.network.connect(
                network=network_name,
                container=self.config.db_container_name,
                alias=db_alias,
            )
        except Exception as ex:
            # already connected or harmless -> ignore
            if not "already exists" in str(ex):
                if self.config.debug:
                    print(f"network connect hint: {ex}", file=sys.stderr)

    def start(self, forceRebuild: bool = False, withInitDB=True):
        """
        start my containers

        Args:
            forceRebuild (bool): if True force rebuilding
            withInitDB (bool): if True intialize my database
        """
        self.up(forceRebuild=forceRebuild)
        if self.config.has_external_db:
            self.prepare_external_db_access()
        if withInitDB:
            msg = "Initializing MediaWiki SQL tables"
            if self.config.has_external_db:
                msg += " and permissions"
            if self.config.verbose:
                print(f"{msg} ...")
            if self.config.has_external_db:
                # Grant permissions first for external DB
                self.execute(
                    "bash",
                    "/scripts/setup-mediawiki.sh",
                    # do not export root password here - try using ENV variable
                    # "--mysql-root-password", self.config.mySQLRootPassword,
                    "--grant",
                )
            dbStatus = self.checkDBConnection()
            if dbStatus.ok:
                # run the mediawiki setup including composer based extensions
                self.setupMediaWiki()
        if self.config.verbose:
            print(
                f"MediaWiki {self.config.container_base_name} is ready at {self.config.full_url}"
            )

    def setupMediaWiki(self):
        """
        setup MediaWiki via the generated script with explicit args
        """
        self.execute(
            "bash",
            "/scripts/setup-mediawiki.sh",
            "--script-dir",
            "/scripts",
            "--web-dir",
            "/var/www/html",
            "--all",
        )

__init__(config)

Constructor

Parameters:

Name	Type	Description	Default
config	MwClusterConfig	MwClusterConfig,	required
home		the home directory to use	required

Source code in mwdocker/docker.py

def __init__(self, config: MwClusterConfig):
    """
    Constructor

    Args:
        config: MwClusterConfig,
        home: the home directory to use
    """
    self.config = config
    # branch as need for git clone e.g. https://gerrit.wikimedia.org/g/mediawiki/extensions/MagicNoCache
    self.branch = f"REL{self.config.getShortVersion('_')}"
    self.composerVersion = 1
    if self.config.shortVersion >= "139":
        self.composerVersion = 2
    # jinja and docker prerequisites
    self.env = self.getJinjaEnv()
    # docker file location
    self.docker_path = (
        f"{self.config.docker_path}/{self.config.container_base_name}"
    )
    os.makedirs(self.docker_path, exist_ok=True)

    self.getContainers()
    self.dbConn = None
    self.wiki_id = self.config.getWikiId()
    self.database = f"{self.wiki_id}_wiki"
    self.dbUser = f"{self.wiki_id}_user"
    self.wikiUser = None
    # Hook to allow modifying results
    # e.g. docker-compose.yaml after generation is finished
    self.postgen_hook = None

check()

check me

Returns:

Name	Type	Description
int	int	exitCode: 0 if ok, 1 if not ok

Source code in mwdocker/docker.py

def check(self) -> int:
    """
    check me

    Returns:
        int: exitCode: 0 if ok, 1 if not ok
    """
    DockerApplication.checkDockerEnvironment(self.config.debug)
    exitCode = 0
    mw, db = self.getContainers()
    if not mw:
        print("mediawiki container missing")
        exitCode = 1
    if not db:
        print("database container missing")
        exitCode = 1
    if mw and db and mw.check() and db.check():
        host_port = mw.getHostPort(80)
        if host_port:
            Logger.check_and_log_equal(
                f"port binding", host_port, "expected  port", str(self.config.port)
            )
            version_url = self.get_version_url(host_port)
            ok = self.checkWiki(version_url)
            if not ok:
                exitCode = 1
        else:
            self.log(f"port binding for port 80 missing", False)
            exitCode = 1
        pass
    return exitCode

checkDBConnection(timeout=10, initialSleep=4.0, factor=1.5, maxTries=9)

check database connection with retries

Parameters:

Name	Type	Description	Default
timeout	float	number of seconds for timeout	10
initialSleep	float	number of seconds to initially wait/sleep	4.0
maxTries	int	maximum number of retries before giving up between each try a sleep is done that starts	9

Returns:

Name	Type	Description
dbStatus	DBStatus	the status

Source code in mwdocker/docker.py

def checkDBConnection(
    self,
    timeout: float = 10,
    initialSleep: float = 4.0,
    factor=1.5,
    maxTries: int = 9,
) -> DBStatus:
    """
    check database connection with retries

    Args:
        timeout (float): number of seconds for timeout
        initialSleep (float): number of seconds to initially wait/sleep
        maxTries (int): maximum number of retries before giving up between each try a sleep is done that starts
        with 0.5 secs and multiplies on every retry

    Returns:
        dbStatus: the status
    """
    conn_msg = f"SQL-Connection to {self.database} on {self.config.host} port {self.config.sql_port} with user {self.dbUser}"
    dbStatus = DBStatus(attempts=0, ok=False, msg=conn_msg, max_tries=maxTries)
    if self.config.verbose:
        print(
            f"Trying {dbStatus.msg} with max {maxTries} tries and {timeout}s timeout per try - initial sleep {initialSleep}s"
        )
    time.sleep(initialSleep)
    sleep = 2.0
    while not dbStatus.ok and dbStatus.attempts <= maxTries:
        try:
            self.doCheckDBConnection(dbStatus, timeout=timeout)
            if not dbStatus.ok:
                if self.config.verbose:
                    print(
                        f"Connection attempt #{dbStatus.attempts}/{dbStatus.max_tries} failed will retry in {sleep:4.1f} secs"
                    )
                # wait before trying
                time.sleep(sleep)
                sleep = sleep * factor
        except Exception as ex:
            dbStatus.ex = ex
            if self.config.verbose:
                print(
                    f"Connection attempt #{dbStatus.attempts} failed with exception {str(ex)} - will not retry ..."
                )
            if self.config.debug:
                print(traceback.format_exc())
            break
    return dbStatus

checkDockerEnvironment(debug=False) staticmethod

check the docker environment

Parameters:

Name	Type	Description	Default
debug	bool	if True show debug information	False

Returns:

Name	Type	Description
str	str	an error message or None

Source code in mwdocker/docker.py

@staticmethod
def checkDockerEnvironment(debug: bool = False) -> str:
    """
    check the docker environment

    Args:
        debug (bool): if True show debug information

    Returns:
        str: an error message or None
    """
    errMsg = None
    os_path = os.environ["PATH"]
    paths = ["/usr/local/bin"]
    for path in paths:
        if os.path.islink(f"{path}/docker"):
            if not path in os_path:
                os.environ["PATH"] = f"{os_path}{os.pathsep}{path}"
                if debug:
                    print(
                        f"""modified PATH from {os_path} to \n{os.environ["PATH"]}"""
                    )
    if not docker.compose.is_installed():
        errMsg = """docker compose up needs to be working"""

    return errMsg

checkWiki(version_url)

check this wiki against the content of the given version_url

Source code in mwdocker/docker.py

def checkWiki(self, version_url: str) -> bool:
    """
    check this wiki against the content of the given version_url
    """
    print(f"Checking {version_url} ...")
    ok = True
    try:
        html_tables = HtmlTables(version_url)
        tables = html_tables.get_tables("h2")
        if self.config.debug:
            p = pprint.PrettyPrinter(indent=2)
            p.pprint(tables)
        ok = ok and Logger.check_and_log(
            "Special Version accessible ...", "Installed software" in tables
        )
        if ok:
            software = tables["Installed software"]
            software_map, _dup = LOD.getLookup(
                software, "Product", withDuplicates=False
            )
            mw_version = software_map["MediaWiki"]["Version"]
            ok = ok and Logger.check_and_log_equal(
                "Mediawiki Version", mw_version, "expected ", self.config.version
            )
            db_version_str = software_map["MariaDB"]["Version"]
            db_version = MariaDB.getVersion(db_version_str)
            ok = ok and Logger.check_and_log(
                f"Maria DB Version {db_version} fitting expected {self.config.mariaDBVersion}?",
                self.config.mariaDBVersion.startswith(db_version),
            )
            pass
    except Exception as ex:
        ok = Logger.check_and_log(str(ex), False)
    return ok

close()

close the database

Source code in mwdocker/docker.py

def close(self):
    """
    close the database
    """
    self.dbClose()

createOrModifyWikiUser(wikiId, force_overwrite=False, lenient=False)

create or modify the WikiUser for this DockerApplication

Parameters:

Name	Type	Description	Default
wikiId	str	the wikiId to create or modify a wiki user for	required
force_overwrite	bool	if True overwrite the wikiuser info	False
lenient(bool)		do not throw Exception if wikiuser exists	required

Source code in mwdocker/docker.py

def createOrModifyWikiUser(
    self, wikiId, force_overwrite: bool = False, lenient: bool = False
) -> WikiUser:
    """
    create or modify the WikiUser for this DockerApplication

    Args:
        wikiId (str): the wikiId to create or modify a wiki user for
        force_overwrite (bool): if True overwrite the wikiuser info
        lenient(bool): do not throw Exception if wikiuser exists
    """
    wikiUsers = WikiUser.getWikiUsers(lenient=True)
    if wikiId in wikiUsers and not force_overwrite:
        wikiUser = wikiUsers[wikiId]
        if self.config.password != wikiUser.getPassword():
            msg = f"wikiUser for wiki {wikiId} already exists but with different password"
            if lenient:
                print(msg, file=sys.stderr)
            else:
                raise Exception(msg)
        pass
    else:
        wikiUser = self.createWikiUser(wikiId, store=True)
    return wikiUser

createWikiUser(wikiId=None, store=False)

create my wikiUser and optionally save it

Parameters:

Name	Type	Description	Default
store	bool	if True save my user data to the relevant ini File	False

Source code in mwdocker/docker.py

def createWikiUser(self, wikiId: str = None, store: bool = False):
    """
    create my wikiUser and optionally save it

    Args:
       store (bool): if True save my user data to the relevant ini File
    """
    if not wikiId:
        wikiId = f"{self.config.container_base_name}"
    userDict = {
        "wikiId": f"{wikiId}",
        "url": f"{self.config.base_url}:{self.config.port}",
        "scriptPath": f"{self.config.script_path}",
        "user": f"{self.config.user}",
        "email": "noreply@nouser.com",
        "version": f"{self.config.fullVersion}",
        "password": f"{self.config.password}",
    }
    wikiUser = WikiUser.ofDict(userDict, encrypted=False)
    if store:
        wikiUser.save()
    return wikiUser

dbClose()

close the database connection

Source code in mwdocker/docker.py

def dbClose(self):
    """
    close the database connection
    """
    if self.dbConn and self.dbConn.is_connected():
        self.dbConn.close()

dbConnect(timeout=10)

connect to the database and return the connection

Parameters:

Name	Type	Description	Default
timeout	int	number of seconds for timeout	10

Returns:

Type	Description
	the connection

Source code in mwdocker/docker.py

def dbConnect(self, timeout: int = 10):
    """
    connect to the database and return the connection

    Args:
        timeout (int): number of seconds for timeout

    Returns:
        the connection
    """
    if self.dbConn is None:
        try:
            self.dbConn = mysql.connector.connect(
                host=self.config.host,
                database=self.database,
                user=self.dbUser,
                port=self.config.sql_port,
                password=self.config.mySQLPassword,
                connection_timeout=timeout,
            )

        except Error as e:
            errMsg = str(e)
            print(
                f"Connection to {self.database} on {self.config.host} with user {self.dbUser} failed error: {errMsg}"
            )
            if "Access denied" in errMsg:
                raise e
    return self.dbConn

doCheckDBConnection(dbStatus, timeout=10)

check the database connection of this application

Parameters:

Name	Type	Description	Default
timeout	int	how many seconds to wait	10

Returns:

Type	Description
	DBStatus

Source code in mwdocker/docker.py

def doCheckDBConnection(self, dbStatus: DBStatus, timeout: int = 10):
    """
    check the database connection of this application

    Args:
        timeout (int): how many seconds to wait

    Returns:
        DBStatus
    """
    dbStatus.attempts += 1
    self.dbConnect(timeout=timeout)
    if self.dbConn and self.dbConn.is_connected():
        rows = self.sqlQuery("select database();")
        dbStatus.ok = True
        if self.config.verbose:
            print(f"{dbStatus.msg} established database returns: {rows[0]}")

down(forceRebuild=False)

run docker compose down

see https://docs.docker.com/engine/reference/commandline/compose_down/ and https://gabrieldemarmiesse.github.io/python-on-whales/sub-commands/compose/#down

Source code in mwdocker/docker.py

def down(self, forceRebuild: bool = False):
    """
    run docker compose down

    see https://docs.docker.com/engine/reference/commandline/compose_down/
    and https://gabrieldemarmiesse.github.io/python-on-whales/sub-commands/compose/#down

    """
    DockerApplication.checkDockerEnvironment(self.config.debug)
    # change directory so that docker CLI will find the relevant dockerfile and docker-compose.yml
    if self.config.verbose:
        print(
            f"running docker compose down for {self.config.container_base_name} {self.config.version} docker application ..."
        )
    # remember current directory
    cwd = os.getcwd()
    os.chdir(self.docker_path)
    try:
        docker.compose.down(volumes=forceRebuild)
    except DockerException as dex:
        print(
            f"warning: docker compose down failed in {self.docker_path}:{str(dex)}"
        )
        pass
    # switch back to previous current directory
    os.chdir(cwd)

execute(*commands)

Execute the given variable list of command strings inside the MediaWiki container.

Parameters:

Name	Type	Description	Default
commands	str	str - command parts to be executed	()

Source code in mwdocker/docker.py

def execute(self, *commands: str):
    """
    Execute the given variable list of command strings inside the MediaWiki container.

    Args:
        commands: str - command parts to be executed
    """
    if not self.mwContainer:
        mwContainerNameDash = self.getContainerName("mw", "-")
        mwContainerNameUnderscore = self.getContainerName("mw", "_")
        raise Exception(
            f"no mediawiki Container {mwContainerNameDash} or {mwContainerNameUnderscore} for {self.name} "
            f"activated by docker compose\n- you might want to check the separator character used "
            f"for container names for your platform {platform.system()}"
        )
    self.mwContainer.execute(*commands, verbose=self.config.verbose)

genComposerRequire(composerFilePath, overwrite=False)

gen the composer.local.json require file

Parameters:

Name	Type	Description	Default
composerFilePath	str	the name of the file to generate	required

Source code in mwdocker/docker.py

def genComposerRequire(self, composerFilePath, overwrite: bool = False):
    """
    gen the composer.local.json require file

    Args:
        composerFilePath (str): the name of the file to generate
    """
    requireJson = self.getComposerRequire()
    self.optionalWrite(composerFilePath, requireJson, overwrite)

generate(templateName, targetPath, overwrite=False, **kwArgs)

generate file at targetPath using the given templateName

Parameters:

Name	Type	Description	Default
templateName	str	the Jinja2 template to use	required
targetPath	str	the path to the target file	required
overwrite	bool	if True overwrite existing files	False
kwArgs()		generic keyword arguments to pass on to template rendering	required

Source code in mwdocker/docker.py

def generate(
    self, templateName: str, targetPath: str, overwrite: bool = False, **kwArgs
):
    """
    generate file at targetPath using the given templateName

    Args:
        templateName (str): the Jinja2 template to use
        targetPath (str): the path to the target file
        overwrite (bool): if True overwrite existing files
        kwArgs(): generic keyword arguments to pass on to template rendering
    """
    try:
        template = self.env.get_template(templateName)
        timestamp = datetime.datetime.now().isoformat()
        content = template.render(
            mwVersion=self.config.version,
            mariaDBVersion=self.config.mariaDBVersion,
            port=self.config.port,
            sql_port=self.config.sql_port,
            smw_version=self.config.smw_version,
            timestamp=timestamp,
            pmwdVersion=Version.version,
            config=self.config,
            **kwArgs,
        )
        self.optionalWrite(targetPath, content, overwrite)

    except TemplateNotFound:
        print(f"no template {templateName} for {self.config.version}")

generateAll(overwrite=False)

generate all files needed for the docker handling

Parameters:

Name	Type	Description	Default
overwrite	bool	if True overwrite the existing files	False

Source code in mwdocker/docker.py

def generateAll(self, overwrite: bool = False):
    """
    generate all files needed for the docker handling

    Args:
        overwrite (bool): if True overwrite the existing files
    """
    # make sure we have the wiki_id ready
    wiki_id = self.config.getWikiId()
    # we have to configure whether
    # bind mounts or volumes are to be used
    if self.config.bind_mount:
        volume_type = "bind"
        mysql_data = f"/var/lib/mediawiki/mysql/{self.config.shortVersion}"
        wiki_sites = f"/var/www/mediawiki/sites"
    else:
        volume_type = "volume"
        mysql_data = "mysql-data"
        wiki_sites = "wiki-sites"
    # first generate Dockerfile
    # the goal is to get an empty MediaWiki (no LocalSettings/extensions)
    # with composer ready
    self.generate(
        "mwDockerfile",
        f"{self.docker_path}/Dockerfile",
        composerVersion=self.composerVersion,
        volume_type=volume_type,
        overwrite=overwrite,
    )
    # the master setup script
    # this used to be part of Dockerfile but
    # needs to be scripted when we use bind mounts due
    # to docker's poor design of permission and mount handling
    self.generate(
        f"setup-mediawiki.sh",
        f"{self.docker_path}/setup-mediawiki.sh",
        script_dir="/scripts",
        web_dir="/var/www/html",
        overwrite=overwrite,
    )
    # the Docker compose
    # at this stage we will have two containers
    # one for the database and one for the MediaWiki
    # the db container may be optionally an existing database container
    template_name = (
        "mwComposeExternalDB.yml"
        if self.config.has_external_db
        else "mwCompose.yml"
    )
    self.generate(
        template_name,
        f"{self.docker_path}/docker-compose.yml",
        # might be None for ExternalDB case
        mySQLRootPassword=self.config.mySQLRootPassword,
        mySQLPassword=self.config.mySQLPassword,
        container_base_name=self.config.container_base_name,
        db_container_name=self.config.db_container_name,
        wiki_id=wiki_id,
        volume_type=volume_type,
        mysql_data=mysql_data,
        wiki_sites=wiki_sites,
        scripts_dir=self.docker_path,
        uid=self.config.uid,
        gid=self.config.gid,
        overwrite=overwrite,
    )
    # now generate the parts we will use later to
    # create the fully configured wiki
    # first - LocalSettings with references to all extensions
    # secretKey
    # https://www.mediawiki.org/wiki/Manual:$wgSecretKey
    # 64 char random hex
    secretKey = secrets.token_hex(32)
    # upgradeKey
    # https://www.mediawiki.org/wiki/Manual:$wgUpgradeKey
    # 16 char random
    upgradeKey = secrets.token_hex(8)
    self.generate(
        f"mwLocalSettings{self.config.shortVersion}.php",
        f"{self.docker_path}/LocalSettings.php",
        wiki_id=wiki_id,
        mySQLPassword=self.config.mySQLPassword,
        hostname=self.config.host,
        extensions=self.config.extensionMap.values(),
        mwShortVersion=self.config.shortVersion,
        logo=self.config.logo,
        secretKey=secretKey,
        upgradeKey=upgradeKey,
        overwrite=overwrite,
    )
    # the SQL file for initial content
    self.generate(
        f"mwWiki{self.config.shortVersion}.sql",
        f"{self.docker_path}/wiki.sql",
        overwrite=overwrite,
    )
    # a WikiUser for automated access via
    # py-3rdparty mediawiki
    if self.config.random_password:
        self.config.password = self.config.create_random_password(
            length=self.config.password_length
        )
        if self.config.wikiId:
            self.createOrModifyWikiUser(
                self.config.wikiId,
                force_overwrite=self.config.force_user,
                lenient=self.config.lenient,
            )
    self.generate(
        f"addSysopUser.sh",
        f"{self.docker_path}/addSysopUser.sh",
        user=self.config.user,
        password=self.config.password,
        overwrite=overwrite,
    )
    self.generate(
        f"installExtensions.sh",
        f"{self.docker_path}/installExtensions.sh",
        extensions=self.config.extensionMap.values(),
        branch=self.branch,
        overwrite=overwrite,
    )
    self.genComposerRequire(
        f"{self.docker_path}/composer.local.json", overwrite=overwrite
    )
    for file_name in [
        "phpinfo.php",
        "disable_sudo.sh",
        "install_djvu.sh",
        "plantuml.sh",
        "upload.ini",
    ]:
        self.generate(
            f"{file_name}", f"{self.docker_path}/{file_name}", overwrite=overwrite
        )

    # chmod generated scripts that need to be bash callable
    # to be executable on container operating system (linux)
    # when bind mounted or being copied
    for executable in [
        "disable_sudo.sh",
        "install_djvu.sh",
        "plantuml.sh",
        "addSysopUser.sh",
        "installExtensions.sh",
        "setup-mediawiki.sh",
    ]:
        path = os.path.join(self.docker_path, executable)
        if os.path.exists(path):
            st = os.stat(path)
            os.chmod(path, st.st_mode | stat.S_IXUSR | stat.S_IXGRP | stat.S_IXOTH)

    if self.postgen_hook:
        self.postgen_hook(self)

    # remember the configuration we used for generating
    # avoid endless loop - forceRebuilds - we have rebuild already
    forceRebuild = self.config.forceRebuild
    self.config.forceRebuild = False
    self.config.save()
    self.config.forceRebuild = forceRebuild

getComposerRequire()

get the json string for the composer require e.g. composer.local.json

Source code in mwdocker/docker.py

    def getComposerRequire(self):
        """
        get the json string for the composer require e.g. composer.local.json
        """
        requires = []
        for ext in self.config.extensionMap.values():
            # get the composer statement
            if ext.composer:
                requires.append(ext.composer)
        indent = "     "
        delim = "" if len(requires) == 0 else ",\n"
        requireList = ""
        if self.config.smw_version:
            requireList += f'{indent}"mediawiki/semantic-media-wiki": "~{self.config.smw_version}"{delim}'
        for i, require in enumerate(requires):
            delim = "" if i >= len(requires) - 1 else ",\n"
            requireList += f"{indent}{require}{delim}"
        requireJson = f"""{{
  "require": {{
{requireList}
  }}
}}"""
        return requireJson

getContainerName(kind, separator)

get my container Name

Source code in mwdocker/docker.py

def getContainerName(self, kind: str, separator: str):
    """
    get my container Name
    """
    containerName = f"{self.config.container_base_name}{separator}{kind}"
    return containerName

getContainers()

get my containers

Returns:

Type	Description
	Tuple(

Source code in mwdocker/docker.py

def getContainers(self):
    """
    get my containers

    Returns:
        Tuple(
    """
    self.dbContainer = None
    self.mwContainer = None
    containerMap = DockerMap.getContainerMap()
    for separator in ["-", "_"]:
        dbContainerName = self.getContainerName("db", separator)
        mwContainerName = self.getContainerName("mw", separator)
        if dbContainerName in containerMap:
            self.dbContainer = DockerContainer(
                dbContainerName, "database", containerMap[dbContainerName]
            )
        if mwContainerName in containerMap:
            self.mwContainer = DockerContainer(
                mwContainerName, "webserver", containerMap[mwContainerName]
            )
    return self.mwContainer, self.dbContainer

getJinjaEnv()

get a Jinja2 environment

Source code in mwdocker/docker.py

def getJinjaEnv(self):
    """
    get a Jinja2 environment
    """
    scriptdir = os.path.dirname(os.path.realpath(__file__))
    resourcePath = os.path.realpath(f"{scriptdir}/resources")
    template_dir = os.path.realpath(f"{resourcePath}/templates")
    # print(f"jinja template directory is {template_dir}")
    env = Environment(loader=FileSystemLoader(template_dir))
    return env

get_version_url(host_port)

get the Special:Version url

Source code in mwdocker/docker.py

def get_version_url(self, host_port: str) -> str:
    """
    get the Special:Version url
    """
    url = self.config.full_url
    # fix url to local port
    # @TODO isn't this superfluous / has no effect ...?
    url = url.replace(str(self.config.port), host_port)
    version_url = f"{url}/index.php?title=Special:Version"
    return version_url

optionalWrite(targetPath, content, overwrite=False)

optionally Write the modified content to the given targetPath

Parameters:

Name	Type	Description	Default
targetPath	str	the path to write the content to	required
content	str	the content to write	required
overwrite	bool	if True overwrite the existing content	False

Source code in mwdocker/docker.py

def optionalWrite(self, targetPath: str, content: str, overwrite: bool = False):
    """
    optionally Write the modified content to the given targetPath

    Args:
        targetPath (str): the path to write the content to
        content (str): the content to write
        overwrite (bool): if True overwrite the existing content
    """
    if not overwrite and os.path.isfile(targetPath):
        if self.config.verbose:
            print(f"{targetPath} already exists!")
        return
    with open(targetPath, "w", newline="") as targetFile:
        targetFile.write(content)

prepare_external_db_access(network_name='db', db_alias='db')

make sure the external database container can be reached

Source code in mwdocker/docker.py

def prepare_external_db_access(
    self, network_name: str = "db", db_alias: str = "db"
):
    """
    make sure the external database container can be reached
    """
    # ensure network exists
    nets = {n.name for n in docker.network.list()}
    if network_name not in nets:
        docker.network.create(network_name)

    # connect external DB container with alias
    try:
        docker.network.connect(
            network=network_name,
            container=self.config.db_container_name,
            alias=db_alias,
        )
    except Exception as ex:
        # already connected or harmless -> ignore
        if not "already exists" in str(ex):
            if self.config.debug:
                print(f"network connect hint: {ex}", file=sys.stderr)

setupMediaWiki()

setup MediaWiki via the generated script with explicit args

Source code in mwdocker/docker.py

def setupMediaWiki(self):
    """
    setup MediaWiki via the generated script with explicit args
    """
    self.execute(
        "bash",
        "/scripts/setup-mediawiki.sh",
        "--script-dir",
        "/scripts",
        "--web-dir",
        "/var/www/html",
        "--all",
    )

sqlQuery(query)

run the given SQL query

Source code in mwdocker/docker.py

def sqlQuery(self, query):
    """
    run the given SQL query
    """
    if self.dbConn and self.dbConn.is_connected():
        cursor = self.dbConn.cursor()
        cursor.execute(query)
        rows = cursor.fetchall()
        cursor.close()
        return rows
    else:
        if self.config.verbose:
            print(
                f"Connection to {self.database} on {self.config.host} with user {self.dbUser} not established"
            )
        return None

start(forceRebuild=False, withInitDB=True)

start my containers

Parameters:

Name	Type	Description	Default
forceRebuild	bool	if True force rebuilding	False
withInitDB	bool	if True intialize my database	True

Source code in mwdocker/docker.py

def start(self, forceRebuild: bool = False, withInitDB=True):
    """
    start my containers

    Args:
        forceRebuild (bool): if True force rebuilding
        withInitDB (bool): if True intialize my database
    """
    self.up(forceRebuild=forceRebuild)
    if self.config.has_external_db:
        self.prepare_external_db_access()
    if withInitDB:
        msg = "Initializing MediaWiki SQL tables"
        if self.config.has_external_db:
            msg += " and permissions"
        if self.config.verbose:
            print(f"{msg} ...")
        if self.config.has_external_db:
            # Grant permissions first for external DB
            self.execute(
                "bash",
                "/scripts/setup-mediawiki.sh",
                # do not export root password here - try using ENV variable
                # "--mysql-root-password", self.config.mySQLRootPassword,
                "--grant",
            )
        dbStatus = self.checkDBConnection()
        if dbStatus.ok:
            # run the mediawiki setup including composer based extensions
            self.setupMediaWiki()
    if self.config.verbose:
        print(
            f"MediaWiki {self.config.container_base_name} is ready at {self.config.full_url}"
        )

up(forceRebuild=False)

start this docker application

Parameters:

Name	Type	Description	Default
forceRebuild	bool	if true stop and remove the existing containers	False

Source code in mwdocker/docker.py

def up(self, forceRebuild: bool = False):
    """
    start this docker application

    Args:
        forceRebuild (bool): if true stop and remove the existing containers
    """
    DockerApplication.checkDockerEnvironment(self.config.debug)
    if self.config.verbose:
        print(
            f"starting {self.config.container_base_name} {self.config.version} docker application ..."
        )
    if forceRebuild:
        for docker_container in [self.dbContainer, self.mwContainer]:
            if docker_container is not None:
                container = docker_container.container
                try:
                    container_name = container.name
                    if self.config.verbose:
                        print(f"stopping and removing container {container_name}")
                except Exception as container_ex:
                    container = None
                if container:
                    try:
                        container.stop()
                    except Exception as stop_ex:
                        if self.config.verbose:
                            print(f"stop failed with {str(stop_ex)}")
                        pass
                    try:
                        container.remove()
                    except Exception as remove_ex:
                        if self.config.verbose:
                            print(f"removed failed with {str(remove_ex)}")
                        pass
                pass

    # remember current directory
    cwd = os.getcwd()

    # change directory so that docker CLI will find the relevant dockerfile and docker-compose.yml
    os.chdir(self.docker_path)
    # project_config = docker.compose.config()
    if forceRebuild:
        docker.compose.build()
    # run docker compose up
    # this might take a while e.g. downloading
    # run docker compose up
    try:
        docker.compose.up(detach=True, force_recreate=forceRebuild)
    except Exception as de:
        print(f"docker compose up failed in {self.docker_path}")
        raise de
        pass
    # switch back to previous current directory
    os.chdir(cwd)

    # check the startup of both containers
    mw, db = self.getContainers()
    for dc in [mw, db]:
        if dc:
            start_secs = dc.wait_for_state(running=True)
            if self.config.verbose:
                print(f"{dc.name} 🟢 started in {start_secs:.2f}s")
    return mw, db

DockerContainer

helper class for docker container info

Source code in mwdocker/docker.py

class DockerContainer:
    """
    helper class for docker container info
    """

    def __init__(self, name, kind, container):
        """
        constructor
        """
        self.name = name
        self.kind = kind
        self.container = container

    def check(self):
        """
        check the given docker container

        print check message and Return if container is running

        Args:
            dc: the docker container

        Returns:
            bool: True if the container is not None
        """
        ok = self.container.state.running
        msg = f"mediawiki {self.kind} container {self.name}"
        return Logger.check_and_log(msg, ok)

    def detect_crash(self) -> str:
        """
        check that we are still running and get crash details if not

        Returns:
            str: None if running, log if crashed
        """
        logs = None
        try:
            if not self.container.state.running:
                logs = docker.container.logs(self.name)
        except Exception as ex:
            logs = str(ex)
        return logs

    def wait_for_state(
        self, running: bool, interval: float = 0.2, timeout: float = 60.0
    ) -> float:
        """
        Wait until the container reaches the desired running state

        Args:
            running: desired running state (True = wait until started, False = wait until stopped)
            interval: polling interval in seconds
            timeout: max time to wait

        Returns:
            float: Time in seconds it took to reach the desired state

        Raises:
            TimeoutError: if the desired state is not reached within timeout
        """
        start_time = time.time()
        deadline = start_time + timeout
        while time.time() < deadline:
            if self.container.state.running == running:
                return time.time() - start_time
            time.sleep(interval)
        state = "running" if running else "stopped"
        raise TimeoutError(
            f"Container '{self.name}' did not reach state '{state}' within {timeout} seconds"
        )

    def getHostPort(self, local_port: int = 80) -> int:
        """
        get the host port for the given local port

        Args:
            local_port (int): the local port to get the mapping for

        Returns:
            int: the  host port or None
        """
        host_port = None
        pb_dict = self.container.host_config.port_bindings
        p_local = f"{local_port}/tcp"
        if p_local in pb_dict:
            pb = pb_dict[p_local][0]
            host_port = pb.host_port
        return host_port

    def execute(self, *commands: str, verbose: bool = False):
        """
        Execute the given variable list of command strings inside the MediaWiki container.

        Args:
            commands: str - command parts to be executed
        """
        command_list = list(commands)

        if verbose:
            command_line = " ".join(command_list)
            print(f"Executing docker command: {command_line}")

        try:
            # see https://gabrieldemarmiesse.github.io/python-on-whales/user_guide/docker_run/#stream-the-output
            for stream_type, stream_content in docker.execute(
                container=self.container, command=command_list, stream=True
            ):
                decoded_line = stream_content.decode("utf-8", errors="replace")
                target_stream = sys.stderr if stream_type == "stderr" else sys.stdout
                print(decoded_line, end="", file=target_stream)

        except Exception as ex:
            logs = self.detect_crash()
            if logs is not None:
                print(f"{self.name} crashed with log: {logs}")
                raise Exception(f"Container {self.name} crashed during execute: {ex}")
            else:
                raise ex

__init__(name, kind, container)

constructor

Source code in mwdocker/docker.py

def __init__(self, name, kind, container):
    """
    constructor
    """
    self.name = name
    self.kind = kind
    self.container = container

check()

check the given docker container

print check message and Return if container is running

Parameters:

Name	Type	Description	Default
dc		the docker container	required

Returns:

Name	Type	Description
bool		True if the container is not None

Source code in mwdocker/docker.py

def check(self):
    """
    check the given docker container

    print check message and Return if container is running

    Args:
        dc: the docker container

    Returns:
        bool: True if the container is not None
    """
    ok = self.container.state.running
    msg = f"mediawiki {self.kind} container {self.name}"
    return Logger.check_and_log(msg, ok)

detect_crash()

check that we are still running and get crash details if not

Returns:

Name	Type	Description
str	str	None if running, log if crashed

Source code in mwdocker/docker.py

def detect_crash(self) -> str:
    """
    check that we are still running and get crash details if not

    Returns:
        str: None if running, log if crashed
    """
    logs = None
    try:
        if not self.container.state.running:
            logs = docker.container.logs(self.name)
    except Exception as ex:
        logs = str(ex)
    return logs

execute(*commands, verbose=False)

Execute the given variable list of command strings inside the MediaWiki container.

Parameters:

Name	Type	Description	Default
commands	str	str - command parts to be executed	()

Source code in mwdocker/docker.py

def execute(self, *commands: str, verbose: bool = False):
    """
    Execute the given variable list of command strings inside the MediaWiki container.

    Args:
        commands: str - command parts to be executed
    """
    command_list = list(commands)

    if verbose:
        command_line = " ".join(command_list)
        print(f"Executing docker command: {command_line}")

    try:
        # see https://gabrieldemarmiesse.github.io/python-on-whales/user_guide/docker_run/#stream-the-output
        for stream_type, stream_content in docker.execute(
            container=self.container, command=command_list, stream=True
        ):
            decoded_line = stream_content.decode("utf-8", errors="replace")
            target_stream = sys.stderr if stream_type == "stderr" else sys.stdout
            print(decoded_line, end="", file=target_stream)

    except Exception as ex:
        logs = self.detect_crash()
        if logs is not None:
            print(f"{self.name} crashed with log: {logs}")
            raise Exception(f"Container {self.name} crashed during execute: {ex}")
        else:
            raise ex

getHostPort(local_port=80)

get the host port for the given local port

Parameters:

Name	Type	Description	Default
local_port	int	the local port to get the mapping for	80

Returns:

Name	Type	Description
int	int	the host port or None

Source code in mwdocker/docker.py

def getHostPort(self, local_port: int = 80) -> int:
    """
    get the host port for the given local port

    Args:
        local_port (int): the local port to get the mapping for

    Returns:
        int: the  host port or None
    """
    host_port = None
    pb_dict = self.container.host_config.port_bindings
    p_local = f"{local_port}/tcp"
    if p_local in pb_dict:
        pb = pb_dict[p_local][0]
        host_port = pb.host_port
    return host_port

wait_for_state(running, interval=0.2, timeout=60.0)

Wait until the container reaches the desired running state

Parameters:

Name	Type	Description	Default
running	bool	desired running state (True = wait until started, False = wait until stopped)	required
interval	float	polling interval in seconds	0.2
timeout	float	max time to wait	60.0

Returns:

Name	Type	Description
float	float	Time in seconds it took to reach the desired state

Raises:

Type	Description
TimeoutError	if the desired state is not reached within timeout

Source code in mwdocker/docker.py

def wait_for_state(
    self, running: bool, interval: float = 0.2, timeout: float = 60.0
) -> float:
    """
    Wait until the container reaches the desired running state

    Args:
        running: desired running state (True = wait until started, False = wait until stopped)
        interval: polling interval in seconds
        timeout: max time to wait

    Returns:
        float: Time in seconds it took to reach the desired state

    Raises:
        TimeoutError: if the desired state is not reached within timeout
    """
    start_time = time.time()
    deadline = start_time + timeout
    while time.time() < deadline:
        if self.container.state.running == running:
            return time.time() - start_time
        time.sleep(interval)
    state = "running" if running else "stopped"
    raise TimeoutError(
        f"Container '{self.name}' did not reach state '{state}' within {timeout} seconds"
    )

docker_map

Created on 2025-08-21

@author: wf

DockerMap

helper class to convert lists of docker elements to maps for improved lookup functionality

Source code in mwdocker/docker_map.py

class DockerMap:
    """
    helper class to convert lists of docker elements to maps for improved
    lookup functionality
    """

    _container_map = None

    @classmethod
    def getContainer(cls, container_name: str):
        containerMap = DockerMap.getContainerMap()
        if not container_name in containerMap:
            raise ValueError(
                f"container {container_name} is not a valid docker container"
            )
        container = containerMap.get(container_name)
        return container

    @classmethod
    def getEnv(cls, container_name: str) -> Dict[str, str]:
        container = cls.getContainer(container_name)
        env_dict = {}
        env_list = container.config.env
        for key_value in env_list:
            if "=" in key_value:
                key, value = key_value.split("=", 1)
                env_dict[key] = value
        return env_dict

    @staticmethod
    def getContainerMap(force_refresh: bool = True) -> Dict[str, Container]:
        """
        get a cached map/dict of containers by container name

        Args:
            force_refresh: if True, refresh from docker instead of using cache
        """
        if DockerMap._container_map is None or force_refresh:
            DockerMap._container_map = {}
            for container in docker.container.list():
                DockerMap._container_map[container.name] = container
        return DockerMap._container_map

getContainerMap(force_refresh=True) staticmethod

get a cached map/dict of containers by container name

Parameters:

Name	Type	Description	Default
force_refresh	bool	if True, refresh from docker instead of using cache	True

Source code in mwdocker/docker_map.py

@staticmethod
def getContainerMap(force_refresh: bool = True) -> Dict[str, Container]:
    """
    get a cached map/dict of containers by container name

    Args:
        force_refresh: if True, refresh from docker instead of using cache
    """
    if DockerMap._container_map is None or force_refresh:
        DockerMap._container_map = {}
        for container in docker.container.list():
            DockerMap._container_map[container.name] = container
    return DockerMap._container_map

html_table

Created on 2022-10-25

@author: wf

HtmlTables

Bases: WebScrape

HtmlTables extractor

Source code in mwdocker/html_table.py

class HtmlTables(WebScrape):
    """
    HtmlTables extractor
    """

    def __init__(self, url: str, debug=False, showHtml=False):
        """
        Constructor

        url(str): the url to read the tables from
        debug(bool): if True switch on debugging
        showHtml(bool): if True show the HTML retrieved
        """
        super().__init__(debug, showHtml)
        self.soup = super().getSoup(url, showHtml)

    def get_tables(self, header_tag: str = None) -> dict:
        """
        get all tables from my soup as a list of list of dicts

        Args:
            header_tag(str): if set search the table name from the given header tag

        Return:
            dict: the list of list of dicts for all tables

        """
        tables = {}
        for i, table in enumerate(self.soup.find_all("table")):
            fields = []
            table_data = []
            category = None
            for tr in table.find_all("tr", recursive=True):
                for th in tr.find_all("th", recursive=True):
                    if "colspan" in th.attrs:
                        category = th.text
                    else:
                        fields.append(th.text)
            for tr in table.find_all("tr", recursive=True):
                record = {}
                for i, td in enumerate(tr.find_all("td", recursive=True)):
                    record[fields[i]] = td.text
                if record:
                    if category:
                        record["category"] = category
                    table_data.append(record)
            if header_tag is not None:
                header = table.find_previous_sibling(header_tag)
                table_name = header.text
            else:
                table_name = f"table{i}"
            tables[table_name] = table_data
        return tables

__init__(url, debug=False, showHtml=False)

Constructor

url(str): the url to read the tables from debug(bool): if True switch on debugging showHtml(bool): if True show the HTML retrieved

Source code in mwdocker/html_table.py

def __init__(self, url: str, debug=False, showHtml=False):
    """
    Constructor

    url(str): the url to read the tables from
    debug(bool): if True switch on debugging
    showHtml(bool): if True show the HTML retrieved
    """
    super().__init__(debug, showHtml)
    self.soup = super().getSoup(url, showHtml)

get_tables(header_tag=None)

get all tables from my soup as a list of list of dicts

Parameters:

Name	Type	Description	Default
header_tag(str)		if set search the table name from the given header tag	required

Return

dict: the list of list of dicts for all tables

Source code in mwdocker/html_table.py

def get_tables(self, header_tag: str = None) -> dict:
    """
    get all tables from my soup as a list of list of dicts

    Args:
        header_tag(str): if set search the table name from the given header tag

    Return:
        dict: the list of list of dicts for all tables

    """
    tables = {}
    for i, table in enumerate(self.soup.find_all("table")):
        fields = []
        table_data = []
        category = None
        for tr in table.find_all("tr", recursive=True):
            for th in tr.find_all("th", recursive=True):
                if "colspan" in th.attrs:
                    category = th.text
                else:
                    fields.append(th.text)
        for tr in table.find_all("tr", recursive=True):
            record = {}
            for i, td in enumerate(tr.find_all("td", recursive=True)):
                record[fields[i]] = td.text
            if record:
                if category:
                    record["category"] = category
                table_data.append(record)
        if header_tag is not None:
            header = table.find_previous_sibling(header_tag)
            table_name = header.text
        else:
            table_name = f"table{i}"
        tables[table_name] = table_data
    return tables

logger

Created on 2022-10-25

@author: wf

Logger

Bases: object

simple logger

Source code in mwdocker/logger.py

class Logger(object):
    """
    simple logger
    """

    @classmethod
    def check_and_log(cls, msg: str, ok: bool) -> bool:
        """
        log the given message with the given ok flag

        Args:
            msg(str): the message to log/print
            ok(bool): if True show ✅ marker else ❌

        Return:
            bool: the ok parameter for fluid syntax
        """
        marker = "✅" if ok else "❌"
        print(f"{msg}:{marker}")
        return ok

    @classmethod
    def check_and_log_equal(self, nameA, valueA, nameB, valueB):
        msg = f"{nameA} {valueA}= {nameB} {valueB}?"
        return self.check_and_log(msg, valueA == valueB)

check_and_log(msg, ok) classmethod

log the given message with the given ok flag

Parameters:

Name	Type	Description	Default
msg(str)		the message to log/print	required
ok(bool)		if True show ✅ marker else ❌	required

Return

bool: the ok parameter for fluid syntax

Source code in mwdocker/logger.py

@classmethod
def check_and_log(cls, msg: str, ok: bool) -> bool:
    """
    log the given message with the given ok flag

    Args:
        msg(str): the message to log/print
        ok(bool): if True show ✅ marker else ❌

    Return:
        bool: the ok parameter for fluid syntax
    """
    marker = "✅" if ok else "❌"
    print(f"{msg}:{marker}")
    return ok

mariadb

Created on 2022-10-25

@author: wf

MariaDB

Maria DB handling

Source code in mwdocker/mariadb.py

class MariaDB:
    """
    Maria DB handling
    """

    @classmethod
    def getVersion(cls, versionStr: str) -> str:
        """
        get the version from the version String

        Args:
            versionStr(str): the version string to check

        Return:
            str: the extracted version
        """
        # version is anything which is not a dot at beginning
        # two times may be ending with dash
        version_match = re.search(r"([^.]+[.]+[^.-]+)", versionStr)
        version = "?"
        if version_match:
            version = version_match.group(1)
        return version

getVersion(versionStr) classmethod

get the version from the version String

Parameters:

Name	Type	Description	Default
versionStr(str)		the version string to check	required

Return

str: the extracted version

Source code in mwdocker/mariadb.py

@classmethod
def getVersion(cls, versionStr: str) -> str:
    """
    get the version from the version String

    Args:
        versionStr(str): the version string to check

    Return:
        str: the extracted version
    """
    # version is anything which is not a dot at beginning
    # two times may be ending with dash
    version_match = re.search(r"([^.]+[.]+[^.-]+)", versionStr)
    version = "?"
    if version_match:
        version = version_match.group(1)
    return version

mw

Created on 2021-06-23

@author: wf

Extension

represents a MediaWiki extension

Source code in mwdocker/mw.py

@lod_storable
class Extension:
    """
    represents a MediaWiki extension
    """

    name: str
    url: str
    extension: Optional[str] = None
    purpose: Optional[str] = None
    giturl: Optional[str] = None
    composer: Optional[str] = None
    wikidata_id: Optional[str] = None
    since: Optional[str] = None
    localSettings: Optional[str] = None
    require_once_until: Optional[str] = None
    tagmap: Optional[Dict[str, str]] = (
        None  # optionally map MediaWiki REL branches e.g. { "REL1_39": "0.14.0" }
    )

    @classmethod
    def getSamples(cls):
        samplesLOD = [
            {
                "name": "Admin Links",
                "extension": "AdminLinks",
                "url": "https://www.mediawiki.org/wiki/Extension:Admin_Links",
                "purpose": """Admin Links is an extension to MediaWiki that defines a special page, "Special:AdminLinks",
that holds links meant to be helpful for wiki administrators;
it is meant to serve as a "control panel" for the functions an administrator would typically perform in a wiki.
All users can view this page; however, for those with the 'adminlinks' permission (sysops/administrators, by default),
a link to the page also shows up in their "Personal URLs", between "Talk" and "Preferences".""",
                "since": datetime.fromisoformat("2009-05-13"),
                "giturl": "https://gerrit.wikimedia.org/r/mediawiki/extensions/AdminLinks.git",
                "localSettings": "",
            }
        ]
        return samplesLOD

    @classmethod
    def fromSpecialVersionTR(cls, exttr, debug=False):
        """
        Construct an extension from a beautifl soup TR tag
        derived from Special:Version

        Args:
            exttr: the beautiful soup TR tag
            debug(bool): if True show debugging information
        """
        ext = None
        purpose = None
        extNameTag = exttr.find(attrs={"class": "mw-version-ext-name"})
        extPurposeTag = exttr.find(attrs={"class": "mw-version-ext-description"})
        if extNameTag:
            name = extNameTag.string
            extension = name.replace(" ", "")
            url = extNameTag.get("href")
            if extPurposeTag and extPurposeTag.string:
                purpose = extPurposeTag.string
            ext = Extension(name=name, extension=extension, url=url, purpose=purpose)
            ext.getDetailsFromUrl(debug=debug)
        return ext

    def __str__(self):
        text = ""
        delim = ""
        samples = self.getSamples()
        for attr in LOD.getFields(samples):
            if hasattr(self, attr) and getattr(self, attr):
                text += f"{delim}{attr}={getattr(self,attr)}"
                delim = "\n"
        return text

    def getDetailsFromUrl(self, showHtml=False, debug=False):
        """
        get more details from my url
        """
        webscrape = WebScrape()
        try:
            soup = webscrape.getSoup(self.url, showHtml=showHtml)
            for link in soup.findAll("a", attrs={"class": "external text"}):
                if ("GitHub" == link.string) or ("git repository URL") == link.string:
                    self.giturl = link.get("href")
        except urllib.error.HTTPError as herr:
            if debug:
                print(f"HTTPError {str(herr)} for {self.url}")

    def asWikiMarkup(self):
        """
        return me as wiki Markup
        """
        samples = self.getJsonTypeSamples()
        nameValues = ""
        for attr in LOD.getFields(samples):
            if hasattr(self, attr) and getattr(self, attr):
                nameValues += f"|{attr}={getattr(self,attr)}\n"
        wikison = f"""{{{{Extension
{nameValues}
}}}}"""
        return wikison

    def getLocalSettingsLine(self, mwShortVersion: str):
        """
        get my local settings line

        Args:
            mwShortVersion(str): the MediaWiki short version e.g. 127

        Returns:
            entry for LocalSettings
        """
        localSettingsLine = ""
        if self.extension:
            localSettingsLine = f"wfLoadExtension( '{self.extension}' );"
        if self.require_once_until:
            if self.require_once_until >= mwShortVersion:
                localSettingsLine = f'require_once "$IP/extensions/{self.extension}/{self.extension}.php";'

        if self.localSettings:
            localSettingsLine += f"\n  {self.localSettings}"
        return localSettingsLine

    def asScript(self, branch: str = "master") -> str:
        """
        return me as a shell script command line string

        Args:
            branch (str): the MediaWiki branch (e.g. REL1_39)
        """
        script = ""
        if self.giturl:
            options = ""
            # check if tagmap defines a tag for this branch
            tag = None
            if self.tagmap:
                tag = self.tagmap.get(branch)
            if tag:
                # use the mapped tag
                options = f"--branch {tag}"
            elif (
                "//github.com/wikimedia/" in self.giturl
                or "//gerrit.wikimedia.org" in self.giturl
            ):
                # default WMF convention: branch per MediaWiki REL
                options = f"--single-branch --branch {branch}"
            script = f'git_get "{self.giturl}" "{self.extension}" "{options}"'
        else:
            script = "# no installation script command specified"
            if self.composer:
                script += f"\n# installed with composer require {self.composer}"
        return script

asScript(branch='master')

return me as a shell script command line string

Parameters:

Name	Type	Description	Default
branch	str	the MediaWiki branch (e.g. REL1_39)	'master'

Source code in mwdocker/mw.py

def asScript(self, branch: str = "master") -> str:
    """
    return me as a shell script command line string

    Args:
        branch (str): the MediaWiki branch (e.g. REL1_39)
    """
    script = ""
    if self.giturl:
        options = ""
        # check if tagmap defines a tag for this branch
        tag = None
        if self.tagmap:
            tag = self.tagmap.get(branch)
        if tag:
            # use the mapped tag
            options = f"--branch {tag}"
        elif (
            "//github.com/wikimedia/" in self.giturl
            or "//gerrit.wikimedia.org" in self.giturl
        ):
            # default WMF convention: branch per MediaWiki REL
            options = f"--single-branch --branch {branch}"
        script = f'git_get "{self.giturl}" "{self.extension}" "{options}"'
    else:
        script = "# no installation script command specified"
        if self.composer:
            script += f"\n# installed with composer require {self.composer}"
    return script

asWikiMarkup()

return me as wiki Markup

Source code in mwdocker/mw.py

    def asWikiMarkup(self):
        """
        return me as wiki Markup
        """
        samples = self.getJsonTypeSamples()
        nameValues = ""
        for attr in LOD.getFields(samples):
            if hasattr(self, attr) and getattr(self, attr):
                nameValues += f"|{attr}={getattr(self,attr)}\n"
        wikison = f"""{{{{Extension
{nameValues}
}}}}"""
        return wikison

fromSpecialVersionTR(exttr, debug=False) classmethod

Construct an extension from a beautifl soup TR tag derived from Special:Version

Parameters:

Name	Type	Description	Default
exttr		the beautiful soup TR tag	required
debug(bool)		if True show debugging information	required

Source code in mwdocker/mw.py

@classmethod
def fromSpecialVersionTR(cls, exttr, debug=False):
    """
    Construct an extension from a beautifl soup TR tag
    derived from Special:Version

    Args:
        exttr: the beautiful soup TR tag
        debug(bool): if True show debugging information
    """
    ext = None
    purpose = None
    extNameTag = exttr.find(attrs={"class": "mw-version-ext-name"})
    extPurposeTag = exttr.find(attrs={"class": "mw-version-ext-description"})
    if extNameTag:
        name = extNameTag.string
        extension = name.replace(" ", "")
        url = extNameTag.get("href")
        if extPurposeTag and extPurposeTag.string:
            purpose = extPurposeTag.string
        ext = Extension(name=name, extension=extension, url=url, purpose=purpose)
        ext.getDetailsFromUrl(debug=debug)
    return ext

getDetailsFromUrl(showHtml=False, debug=False)

get more details from my url

Source code in mwdocker/mw.py

def getDetailsFromUrl(self, showHtml=False, debug=False):
    """
    get more details from my url
    """
    webscrape = WebScrape()
    try:
        soup = webscrape.getSoup(self.url, showHtml=showHtml)
        for link in soup.findAll("a", attrs={"class": "external text"}):
            if ("GitHub" == link.string) or ("git repository URL") == link.string:
                self.giturl = link.get("href")
    except urllib.error.HTTPError as herr:
        if debug:
            print(f"HTTPError {str(herr)} for {self.url}")

getLocalSettingsLine(mwShortVersion)

get my local settings line

Parameters:

Name	Type	Description	Default
mwShortVersion(str)		the MediaWiki short version e.g. 127	required

Returns:

Type	Description
	entry for LocalSettings

Source code in mwdocker/mw.py

def getLocalSettingsLine(self, mwShortVersion: str):
    """
    get my local settings line

    Args:
        mwShortVersion(str): the MediaWiki short version e.g. 127

    Returns:
        entry for LocalSettings
    """
    localSettingsLine = ""
    if self.extension:
        localSettingsLine = f"wfLoadExtension( '{self.extension}' );"
    if self.require_once_until:
        if self.require_once_until >= mwShortVersion:
            localSettingsLine = f'require_once "$IP/extensions/{self.extension}/{self.extension}.php";'

    if self.localSettings:
        localSettingsLine += f"\n  {self.localSettings}"
    return localSettingsLine

ExtensionList

represents a list of MediaWiki extensions

Source code in mwdocker/mw.py

@lod_storable
class ExtensionList:
    """
    represents a list of MediaWiki extensions
    """

    extensions: List[Extension] = field(default_factory=list)

    @staticmethod
    def storeFilePrefix():
        """
        get my storeFilePrefix

        Returns:
            str: the path to where my stored files (e.g. JSON) should be kept
        """
        scriptdir = os.path.dirname(os.path.realpath(__file__))
        resourcePath = os.path.realpath(f"{scriptdir}/resources")
        storeFilePrefix = f"{resourcePath}/extensions"
        return storeFilePrefix

    @classmethod
    def fromSpecialVersion(
        cls, url: str, excludes=["skin", "editor"], showHtml=False, debug=False
    ):
        """
        get an extension List from the given url

        Args:
            url(str): the Special:Version MediaWiki page to read the information from
            exclude (list): a list of types of extensions to exclude
            showHtml(bool): True if the html code should be printed for debugging
            debug(bool): True if debugging should be active

        Returns:
            ExtensionList: an extension list derived from the url
        """
        webscrape = WebScrape()
        soup = webscrape.getSoup(url, showHtml=showHtml)

        # search for
        # <tr class="mw-version-ext" id="mw-version-ext-media-PDF_Handler">
        exttrs = soup.findAll(attrs={"class": "mw-version-ext"})
        extList = ExtensionList()
        for exttr in exttrs:
            if showHtml:
                print(exttr)
            doExclude = False
            for exclude in excludes:
                if f"-{exclude}-" in exttr.get("id"):
                    doExclude = True
            if not doExclude:
                ext = Extension.fromSpecialVersionTR(exttr, debug=debug)
                if ext:
                    extList.extensions.append(ext)
        return extList

    @classmethod
    def restore(cls) -> "ExtensionList":
        """
        restore the extension list
        """
        path = ExtensionList.storeFilePrefix()
        yaml_file = f"{path}.yaml"
        extlist = ExtensionList.load_from_yaml_file(yaml_file)
        return extlist

fromSpecialVersion(url, excludes=['skin', 'editor'], showHtml=False, debug=False) classmethod

get an extension List from the given url

Parameters:

Name	Type	Description	Default
url(str)		the Special:Version MediaWiki page to read the information from	required
exclude	list	a list of types of extensions to exclude	required
showHtml(bool)		True if the html code should be printed for debugging	required
debug(bool)		True if debugging should be active	required

Returns:

Name	Type	Description
ExtensionList		an extension list derived from the url

Source code in mwdocker/mw.py

@classmethod
def fromSpecialVersion(
    cls, url: str, excludes=["skin", "editor"], showHtml=False, debug=False
):
    """
    get an extension List from the given url

    Args:
        url(str): the Special:Version MediaWiki page to read the information from
        exclude (list): a list of types of extensions to exclude
        showHtml(bool): True if the html code should be printed for debugging
        debug(bool): True if debugging should be active

    Returns:
        ExtensionList: an extension list derived from the url
    """
    webscrape = WebScrape()
    soup = webscrape.getSoup(url, showHtml=showHtml)

    # search for
    # <tr class="mw-version-ext" id="mw-version-ext-media-PDF_Handler">
    exttrs = soup.findAll(attrs={"class": "mw-version-ext"})
    extList = ExtensionList()
    for exttr in exttrs:
        if showHtml:
            print(exttr)
        doExclude = False
        for exclude in excludes:
            if f"-{exclude}-" in exttr.get("id"):
                doExclude = True
        if not doExclude:
            ext = Extension.fromSpecialVersionTR(exttr, debug=debug)
            if ext:
                extList.extensions.append(ext)
    return extList

restore() classmethod

restore the extension list

Source code in mwdocker/mw.py

@classmethod
def restore(cls) -> "ExtensionList":
    """
    restore the extension list
    """
    path = ExtensionList.storeFilePrefix()
    yaml_file = f"{path}.yaml"
    extlist = ExtensionList.load_from_yaml_file(yaml_file)
    return extlist

storeFilePrefix() staticmethod

get my storeFilePrefix

Returns:

Name	Type	Description
str		the path to where my stored files (e.g. JSON) should be kept

Source code in mwdocker/mw.py

@staticmethod
def storeFilePrefix():
    """
    get my storeFilePrefix

    Returns:
        str: the path to where my stored files (e.g. JSON) should be kept
    """
    scriptdir = os.path.dirname(os.path.realpath(__file__))
    resourcePath = os.path.realpath(f"{scriptdir}/resources")
    storeFilePrefix = f"{resourcePath}/extensions"
    return storeFilePrefix

mwcluster

Created on 2021-08-06 @author: wf

MediaWikiCluster

Bases: object

a cluster of mediawiki docker Applications

Source code in mwdocker/mwcluster.py

class MediaWikiCluster(object):
    """
    a cluster of mediawiki docker Applications
    """

    # https://hub.docker.com/_/mediawiki
    # 2023-01-13
    # MediaWiki Extensions and Skins Security Release Supplement (1.35.9/1.38.4/1.39.1)
    # 2023-02-23 1.39.2 released
    # 2023-04-04 1.39.3 upgrade
    # 2023-10-04 1.39.5 upgrade
    # 2024-04-15 1.39.7 upgrade
    # 2024-08-02 1.39.8 upgrade
    # 2024-10-11 1.39.10 upgrade
    # 2025-03-18 1.39.11 upgrade 1.43.0 addition
    # 2025-04-13 Security and maintenance release: 1.39.12 / 1.42.6 / 1.43.1
    # 2025-06-30 Security and maintenance release: 1.39.13 / 1.42.7 / 1.43.3
    # 2025-10-05 Security and maintenance release: 1.39.14 / 1.43.4 / 1.44.1
    # 2025-11-14 Security and maintenance release: 1.39.15 / 1.43.5 / 1.44.2
    # 2025-12-18 Security and maintenance release: 1.39.16 / 1.43.6 / 1.44.3 / 1.45.1
    # 2025-12-18 1.39.17 is also out in docker images

    def __init__(self, config: MwClusterConfig, args: Namespace = None):
        """
        Constructor

        Args:
            config(MWClusterConfig): the MediaWiki Cluster Configuration to use
        """
        self.config = config
        self.args = args
        self.apps = {}

    def createApps(self, withGenerate: bool = True) -> dict:
        """
        create my apps

        Args:
            withGenerate(bool): if True generate the config files

        Returns:
            dict(str): a dict of apps by version
        """
        exitCode = self.checkDocker()
        if exitCode > 0:
            raise ValueError("createApps needs docker command in PATH")
        app_count = len(self.config.versions)
        for i, version in enumerate(self.config.versions):
            mwApp = self.getDockerApplication(i, app_count, version)
            if withGenerate:
                mwApp.generateAll(overwrite=self.config.forceRebuild)
            self.apps[version] = mwApp
        return self.apps

    def checkDocker(self) -> int:
        """
        check the Docker environment

        print an error message on stderr if check fails

        Returns:
            int: exitCode - 0 if ok 1 if failed

        """
        errMsg = DockerApplication.checkDockerEnvironment(self.config.debug)
        if errMsg is not None:
            print(errMsg, file=sys.stderr)
            return 1
        return 0

    def start(self, forceRebuild: bool = False, withInitDB=True) -> int:
        """
        create and start the composer applications

        Returns:
            int: exitCode - 0 if ok 1 if failed
        """
        exitCode = self.checkDocker()
        if exitCode > 0:
            return exitCode

        for version in self.config.versions:
            mwApp = self.apps[version]
            mwApp.start(forceRebuild=forceRebuild, withInitDB=withInitDB)
        return 0

    def down(self, forceRebuild: bool = False):
        """
        run docker compose down
        """
        exitCode = self.checkDocker()
        if exitCode > 0:
            return exitCode
        for _i, version in enumerate(self.config.versions):
            mwApp = self.apps[version]
            mwApp.down(forceRebuild)

    def listWikis(self) -> int:
        """
        list my wikis

        Returns:
            int: exitCode - 0 if ok 1 if failed
        """
        exitCode = self.checkDocker()
        if exitCode > 0:
            return exitCode
        for i, version in enumerate(self.config.versions):
            mwApp = self.apps[version]
            mw, db = mwApp.getContainers()
            config = mwApp.config
            ok = mw and db
            msg = f"{i+1}:{config.container_base_name} {config.fullVersion}"
            Logger.check_and_log(msg, ok)
        return exitCode

    def check(self) -> int:
        """
        check the composer applications

        Returns:
            int: exitCode - 0 if ok 1 if failed
        """
        exitCode = self.checkDocker()
        if exitCode > 0:
            return exitCode

        for i, version in enumerate(self.config.versions):
            mwApp = self.apps[version]
            msg = f"{i+1}:checking {version} ..."
            print(msg)
            exitCode = mwApp.check()
        return exitCode

    def close(self):
        """
        close my apps
        """
        for mwApp in self.apps.values():
            mwApp.close()

    def getDockerApplication(self, i: int, count: int, version: str):
        """
        get the docker application for the given version index and version

        Args:
            i(int): the index of the version
            count(int): total number of Docker applications in this cluster
            version(str): the mediawiki version to use

        Returns:
            DockerApplication: the docker application
        """
        # please note that we are using the subclass MwClusterConfig here although
        # we only need the superclass MwConfig - we let inheritance work here for us but
        # have to ignore the superfluous fields
        appConfig = dataclasses.replace(self.config)
        appConfig.extensionMap = self.config.extensionMap.copy()
        appConfig.version = version
        appConfig.base_port = self.config.base_port + i
        appConfig.port = self.config.base_port + i
        appConfig.sql_port = self.config.sql_port + i
        # let post_init create a new container_base_name and db_container_name
        if count > 1:
            appConfig.container_base_name = None
            appConfig.db_container_name = (
                self.args.db_container_name if self.args else None
            )
        appConfig.__post_init__()
        mwApp = DockerApplication(config=appConfig)
        return mwApp

__init__(config, args=None)

Constructor

Parameters:

Name	Type	Description	Default
config(MWClusterConfig)		the MediaWiki Cluster Configuration to use	required

Source code in mwdocker/mwcluster.py

def __init__(self, config: MwClusterConfig, args: Namespace = None):
    """
    Constructor

    Args:
        config(MWClusterConfig): the MediaWiki Cluster Configuration to use
    """
    self.config = config
    self.args = args
    self.apps = {}

check()

check the composer applications

Returns:

Name	Type	Description
int	int	exitCode - 0 if ok 1 if failed

Source code in mwdocker/mwcluster.py

def check(self) -> int:
    """
    check the composer applications

    Returns:
        int: exitCode - 0 if ok 1 if failed
    """
    exitCode = self.checkDocker()
    if exitCode > 0:
        return exitCode

    for i, version in enumerate(self.config.versions):
        mwApp = self.apps[version]
        msg = f"{i+1}:checking {version} ..."
        print(msg)
        exitCode = mwApp.check()
    return exitCode

checkDocker()

check the Docker environment

print an error message on stderr if check fails

Returns:

Name	Type	Description
int	int	exitCode - 0 if ok 1 if failed

Source code in mwdocker/mwcluster.py

def checkDocker(self) -> int:
    """
    check the Docker environment

    print an error message on stderr if check fails

    Returns:
        int: exitCode - 0 if ok 1 if failed

    """
    errMsg = DockerApplication.checkDockerEnvironment(self.config.debug)
    if errMsg is not None:
        print(errMsg, file=sys.stderr)
        return 1
    return 0

close()

close my apps

Source code in mwdocker/mwcluster.py

def close(self):
    """
    close my apps
    """
    for mwApp in self.apps.values():
        mwApp.close()

createApps(withGenerate=True)

create my apps

Parameters:

Name	Type	Description	Default
withGenerate(bool)		if True generate the config files	required

Returns:

Name	Type	Description
dict	str	a dict of apps by version

Source code in mwdocker/mwcluster.py

def createApps(self, withGenerate: bool = True) -> dict:
    """
    create my apps

    Args:
        withGenerate(bool): if True generate the config files

    Returns:
        dict(str): a dict of apps by version
    """
    exitCode = self.checkDocker()
    if exitCode > 0:
        raise ValueError("createApps needs docker command in PATH")
    app_count = len(self.config.versions)
    for i, version in enumerate(self.config.versions):
        mwApp = self.getDockerApplication(i, app_count, version)
        if withGenerate:
            mwApp.generateAll(overwrite=self.config.forceRebuild)
        self.apps[version] = mwApp
    return self.apps

down(forceRebuild=False)

run docker compose down

Source code in mwdocker/mwcluster.py

def down(self, forceRebuild: bool = False):
    """
    run docker compose down
    """
    exitCode = self.checkDocker()
    if exitCode > 0:
        return exitCode
    for _i, version in enumerate(self.config.versions):
        mwApp = self.apps[version]
        mwApp.down(forceRebuild)

getDockerApplication(i, count, version)

get the docker application for the given version index and version

Parameters:

Name	Type	Description	Default
i(int)		the index of the version	required
count(int)		total number of Docker applications in this cluster	required
version(str)		the mediawiki version to use	required

Returns:

Name	Type	Description
DockerApplication		the docker application

Source code in mwdocker/mwcluster.py

def getDockerApplication(self, i: int, count: int, version: str):
    """
    get the docker application for the given version index and version

    Args:
        i(int): the index of the version
        count(int): total number of Docker applications in this cluster
        version(str): the mediawiki version to use

    Returns:
        DockerApplication: the docker application
    """
    # please note that we are using the subclass MwClusterConfig here although
    # we only need the superclass MwConfig - we let inheritance work here for us but
    # have to ignore the superfluous fields
    appConfig = dataclasses.replace(self.config)
    appConfig.extensionMap = self.config.extensionMap.copy()
    appConfig.version = version
    appConfig.base_port = self.config.base_port + i
    appConfig.port = self.config.base_port + i
    appConfig.sql_port = self.config.sql_port + i
    # let post_init create a new container_base_name and db_container_name
    if count > 1:
        appConfig.container_base_name = None
        appConfig.db_container_name = (
            self.args.db_container_name if self.args else None
        )
    appConfig.__post_init__()
    mwApp = DockerApplication(config=appConfig)
    return mwApp

listWikis()

list my wikis

Returns:

Name	Type	Description
int	int	exitCode - 0 if ok 1 if failed

Source code in mwdocker/mwcluster.py

def listWikis(self) -> int:
    """
    list my wikis

    Returns:
        int: exitCode - 0 if ok 1 if failed
    """
    exitCode = self.checkDocker()
    if exitCode > 0:
        return exitCode
    for i, version in enumerate(self.config.versions):
        mwApp = self.apps[version]
        mw, db = mwApp.getContainers()
        config = mwApp.config
        ok = mw and db
        msg = f"{i+1}:{config.container_base_name} {config.fullVersion}"
        Logger.check_and_log(msg, ok)
    return exitCode

start(forceRebuild=False, withInitDB=True)

create and start the composer applications

Returns:

Name	Type	Description
int	int	exitCode - 0 if ok 1 if failed

Source code in mwdocker/mwcluster.py

def start(self, forceRebuild: bool = False, withInitDB=True) -> int:
    """
    create and start the composer applications

    Returns:
        int: exitCode - 0 if ok 1 if failed
    """
    exitCode = self.checkDocker()
    if exitCode > 0:
        return exitCode

    for version in self.config.versions:
        mwApp = self.apps[version]
        mwApp.start(forceRebuild=forceRebuild, withInitDB=withInitDB)
    return 0

mwdocker_cmd

Created on 2025-08-01

@author: wf

MediaWikiDockerCmd

Bases: BaseCmd

pymediawiki docker main

Source code in mwdocker/mwdocker_cmd.py

class MediaWikiDockerCmd(BaseCmd):
    """
    pymediawiki docker main
    """

    def __init__(self, version=Version):
        super().__init__(version)
        self.config = MwClusterConfig()
        self.cluster = None

    def getMwConfig(self, argv=None, version=None) -> MwClusterConfig:
        """
        get a mediawiki configuration for the given command line arguments
        """
        if not argv:
            argv = self.argv
        parser = ArgumentParser()
        if version is None:
            version = MwConfig.version
        mwClusterConfig = MwClusterConfig(version=version)
        self.config = mwClusterConfig
        self.add_arguments(parser)
        args = parser.parse_args(argv)
        mwClusterConfig.fromArgs(args)
        return mwClusterConfig

    def add_arguments(self, parser: ArgumentParser):
        """
        add parser arguments
        """
        super().add_arguments(parser)
        self.config.addArgs(parser)
        parser.add_argument("--create", action="store_true")
        parser.add_argument("--down", action="store_true")
        parser.add_argument("--check", action="store_true")
        parser.add_argument("--list", action="store_true")

    def handle_args(self, args: Namespace) -> bool:
        if super().handle_args(args):
            return True
        self.config.fromArgs(args)
        self.cluster = MediaWikiCluster(self.config, args)
        self.cluster.createApps(withGenerate=args.create)
        if args.check:
            self.exit_code = self.cluster.check()
        elif args.create:
            self.exit_code = self.cluster.start(forceRebuild=self.config.forceRebuild)
        elif args.list:
            self.exit_code = self.cluster.listWikis()
        elif args.down:
            self.exit_code = self.cluster.down(forceRebuild=self.config.forceRebuild)
        else:
            self.parser.print_usage()
            self.exit_code = 1
        return True

add_arguments(parser)

add parser arguments

Source code in mwdocker/mwdocker_cmd.py

def add_arguments(self, parser: ArgumentParser):
    """
    add parser arguments
    """
    super().add_arguments(parser)
    self.config.addArgs(parser)
    parser.add_argument("--create", action="store_true")
    parser.add_argument("--down", action="store_true")
    parser.add_argument("--check", action="store_true")
    parser.add_argument("--list", action="store_true")

getMwConfig(argv=None, version=None)

get a mediawiki configuration for the given command line arguments

Source code in mwdocker/mwdocker_cmd.py

def getMwConfig(self, argv=None, version=None) -> MwClusterConfig:
    """
    get a mediawiki configuration for the given command line arguments
    """
    if not argv:
        argv = self.argv
    parser = ArgumentParser()
    if version is None:
        version = MwConfig.version
    mwClusterConfig = MwClusterConfig(version=version)
    self.config = mwClusterConfig
    self.add_arguments(parser)
    args = parser.parse_args(argv)
    mwClusterConfig.fromArgs(args)
    return mwClusterConfig

version

Created on 2022-04-07

@author: wf

Version

Bases: object

Version handling for pymediawikidocker

Source code in mwdocker/version.py

class Version(object):
    """
    Version handling for pymediawikidocker
    """

    name = "pymediawikidocker"
    version = mwdocker.__version__
    date = "2021-06-21"
    updated = "2025-12-16"

    authors = "Wolfgang Fahl, Tim Holzheim"

    description = (
        "Python controlled (semantic) mediawiki docker application cluster installation"
    )

    cm_url = "https://github.com/WolfgangFahl/pymediawikidocker"
    chat_url = "https://github.com/WolfgangFahl/pymediawikidocker/discussions"
    doc_url = "https://wiki.bitplan.com/index.php/Pymediawikidocker"

    license = f"""Copyright 2020-2025 contributors. All rights reserved.
  Licensed under the Apache License 2.0
  http://www.apache.org/licenses/LICENSE-2.0
  Distributed on an "AS IS" basis without warranties
  or conditions of any kind, either express or implied."""
    longDescription = f"""{name} version {version}
{description}
  Created by {authors} on {date} last updated {updated}"""

webscrape

Created on 2020-08-20

@author: wf

WebScrape

Bases: object

WebScraper

Source code in mwdocker/webscrape.py

class WebScrape(object):
    """
    WebScraper
    """

    def __init__(self, debug=False, showHtml=False):
        """
        Constructor
        """
        self.err = None
        self.valid = False
        self.debug = debug
        self.showHtml = showHtml

    def getSoup(self, url, showHtml):
        """
        get the beautiful Soup parser

        Args:
           showHtml(boolean): True if the html code should be pretty printed and shown
        """
        req = Request(url, headers={"User-Agent": "Mozilla/5.0"})
        html = urlopen(req).read()
        soup = BeautifulSoup(html, "html.parser", from_encoding="utf-8")
        if showHtml:
            self.printPrettyHtml(soup)

        return soup

    def printPrettyHtml(self, soup):
        """
        print the prettified html for the given soup

        Args:
            soup(BeuatifulSoup): the parsed html to print
        """
        prettyHtml = soup.prettify()
        print(prettyHtml)

__init__(debug=False, showHtml=False)

Constructor

Source code in mwdocker/webscrape.py

def __init__(self, debug=False, showHtml=False):
    """
    Constructor
    """
    self.err = None
    self.valid = False
    self.debug = debug
    self.showHtml = showHtml

getSoup(url, showHtml)

get the beautiful Soup parser

Parameters:

Name	Type	Description	Default
showHtml(boolean)		True if the html code should be pretty printed and shown	required

Source code in mwdocker/webscrape.py

def getSoup(self, url, showHtml):
    """
    get the beautiful Soup parser

    Args:
       showHtml(boolean): True if the html code should be pretty printed and shown
    """
    req = Request(url, headers={"User-Agent": "Mozilla/5.0"})
    html = urlopen(req).read()
    soup = BeautifulSoup(html, "html.parser", from_encoding="utf-8")
    if showHtml:
        self.printPrettyHtml(soup)

    return soup

printPrettyHtml(soup)

print the prettified html for the given soup

Parameters:

Name	Type	Description	Default
soup(BeuatifulSoup)		the parsed html to print	required

Source code in mwdocker/webscrape.py

def printPrettyHtml(self, soup):
    """
    print the prettified html for the given soup

    Args:
        soup(BeuatifulSoup): the parsed html to print
    """
    prettyHtml = soup.prettify()
    print(prettyHtml)