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 the fully qualifying hostname
        of the computer the server runs on

        Returns:
            str: the hostname
        """
        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 = "localhost"  # host="127.0.0.1"
        return host

get_default_host() classmethod

get the default host as the fully qualifying hostname of the computer the server runs on

Returns:

Name	Type	Description
str	str	the hostname

Source code in mwdocker/config.py

@classmethod
def get_default_host(cls) -> str:
    """
    get the default host as the fully qualifying hostname
    of the computer the server runs on

    Returns:
        str: the hostname
    """
    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 = "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.38.6", "1.39.8", "1.40.4", "1.41.2","1.42.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] ",
        )

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] ",
    )

MwConfig dataclass

MediaWiki configuration for a Single Wiki

Source code in mwdocker/config.py

@dataclass
class MwConfig:
    """
    MediaWiki configuration for a Single Wiki
    """

    version: str = "1.39.8"
    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"
    password_length: int = 15
    random_password: bool = False
    force_user: bool = False
    password: str = "sysop-1234!"
    mySQLRootPassword: Optional[str] = None
    mySQLPassword: Optional[str] = None
    logo: str = "$wgResourceBasePath/resources/assets/wiki.png"
    port: int = 9080
    sql_port: int = 9306
    url = None
    full_url = None
    prot: str = "http"
    host: str = Host.get_default_host()
    script_path: str = ""
    container_base_name: str = None
    networkName: str = "mwNetwork"
    mariaDBVersion: str = "10.11"
    forceRebuild: bool = False
    debug: bool = False
    verbose: bool = True
    wikiId: Optional[str] = None
    docker_path: Optional[str] = None

    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}"
        self.reset_url(self.url)

    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.full_url = f"{self.base_url}{self.script_path}:{self.port}"

    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()
        print(json_str, file=open(path, "w"))
        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_dict = json.loads(json_str)
            config = dacite.from_dict(data_class=self.__class__, data=config_dict)
            # 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("(?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
        """
        return secrets.token_urlsafe(length)

    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()
        if extensionJsonFile is not None:
            extraExtensionList = ExtensionList.load_from_json_file(extensionJsonFile)
            for ext in extraExtensionList.extensions:
                extensionList.extensions.append(ext)
        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 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.container_base_name = args.container_name
        self.docker_path = args.docker_path
        self.extensionNameList = args.extensionNameList
        self.extensionJsonFile = args.extensionJsonFile
        self.forceRebuild = args.forceRebuild
        self.host = args.host
        self.logo = args.logo
        self.mariaDBVersion = args.mariaDBVersion
        # passwords
        self.mySQLRootPassword = args.mysqlPassword
        if not self.mySQLRootPassword:
            self.mySQLRootPassword = self.create_random_password(self.password_length)
        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.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(
            "-cn",
            "--container_name",
            default=self.container_base_name,
            help="set container name (only valid and recommended for single version call)",
        )
        parser.add_argument(
            "-d",
            "--debug",
            dest="debug",
            action="store_true",
            default=self.debug,
            help="enable debug mode [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]",
        )
        parser.add_argument(
            "-f",
            "--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(
            "--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(
            "--mysqlPassword",
            default=self.mySQLRootPassword,
            help="set sqlRootPassword [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(
            "-q",
            "--quiet",
            default=not self.verbose,
            help="not verbose [default: %(default)s]",
            action="store_true",
        )

__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}"
    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(
        "-cn",
        "--container_name",
        default=self.container_base_name,
        help="set container name (only valid and recommended for single version call)",
    )
    parser.add_argument(
        "-d",
        "--debug",
        dest="debug",
        action="store_true",
        default=self.debug,
        help="enable debug mode [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]",
    )
    parser.add_argument(
        "-f",
        "--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(
        "--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(
        "--mysqlPassword",
        default=self.mySQLRootPassword,
        help="set sqlRootPassword [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(
        "-q",
        "--quiet",
        default=not self.verbose,
        help="not verbose [default: %(default)s]",
        action="store_true",
    )

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
    """
    return secrets.token_urlsafe(length)

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.container_base_name = args.container_name
    self.docker_path = args.docker_path
    self.extensionNameList = args.extensionNameList
    self.extensionJsonFile = args.extensionJsonFile
    self.forceRebuild = args.forceRebuild
    self.host = args.host
    self.logo = args.logo
    self.mariaDBVersion = args.mariaDBVersion
    # passwords
    self.mySQLRootPassword = args.mysqlPassword
    if not self.mySQLRootPassword:
        self.mySQLRootPassword = self.create_random_password(self.password_length)
    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.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	required
extensionJsonFile(str)		the name of an extra extensionJsonFile (if any)	required

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()
    if extensionJsonFile is not None:
        extraExtensionList = ExtensionList.load_from_json_file(extensionJsonFile)
        for ext in extraExtensionList.extensions:
            extensionList.extensions.append(ext)
    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 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("(?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_dict = json.loads(json_str)
        config = dacite.from_dict(data_class=self.__class__, data=config_dict)
        # 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.full_url = f"{self.base_url}{self.script_path}:{self.port}"

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()
    print(json_str, file=open(path, "w"))
    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.database = "wiki"
        self.dbUser = "wikiuser"
        self.wikiUser = 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
        if not docker.compose.is_installed():
            errMsg = """docker compose up needs to be working"""
        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"]}"""
                        )
        return errMsg

    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)
                )
                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/Special:Version"

                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 initDB(self):
        """
        initialize my SQL database
        """
        # restore the mySQL dump data
        self.execute("/root/initdb.sh")
        # update the database e.g. to initialize Semantic MediaWiki tables
        self.execute("/root/update.sh")
        # add an initial sysop user as specified
        self.execute("/root/addSysopUser.sh")

    def installExtensions(self):
        """
        install all extensions
        """
        self.execute("/root/installExtensions.sh")
        self.execute("/root/fixPermissions.sh")

    def startUp(self):
        """
        run startUp scripts
        """
        # fix permissions
        self.execute("/root/fixPermissions.sh")
        # add language icons
        self.execute("/root/lang.sh", "--site", "/var/www/html")
        # start cron job
        self.execute("/root/addCronTabEntry.sh")

    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) -> 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
        """
        wikiUsers = WikiUser.getWikiUsers(lenient=True)
        if wikiId in wikiUsers and not force_overwrite:
            wikiUser = wikiUsers[wikiId]
            if self.config.password != wikiUser.getPassword():
                raise Exception(
                    f"wikiUser for wiki {wikiId} already exists but with different password"
                )
            pass
        else:
            wikiUser = self.createWikiUser(wikiId, store=True)
        return wikiUser

    def execute(self, *commands: str):
        """
        execute the given variable list of command strings

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

    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,
                **kwArgs,
            )
            self.optionalWrite(targetPath, content, overwrite)

        except TemplateNotFound:
            print(
                f"no template {templateName} for {self.config.name} {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
        """
        # then generate
        self.generate(
            "mwDockerfile",
            f"{self.docker_path}/Dockerfile",
            composerVersion=self.composerVersion,
            overwrite=overwrite,
        )
        self.generate(
            "mwCompose.yml",
            f"{self.docker_path}/docker-compose.yml",
            mySQLRootPassword=self.config.mySQLRootPassword,
            mySQLPassword=self.config.mySQLPassword,
            container_base_name=self.config.container_base_name,
            overwrite=overwrite,
        )
        self.generate(
            f"mwLocalSettings{self.config.shortVersion}.php",
            f"{self.docker_path}/LocalSettings.php",
            mySQLPassword=self.config.mySQLPassword,
            hostname=self.config.host,
            extensions=self.config.extensionMap.values(),
            mwShortVersion=self.config.shortVersion,
            logo=self.config.logo,
            overwrite=overwrite,
        )
        self.generate(
            f"mwWiki{self.config.shortVersion}.sql",
            f"{self.docker_path}/wiki.sql",
            overwrite=overwrite,
        )
        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
                )
        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 [
            "addCronTabEntry.sh",
            "fixPermissions.sh",
            "initdb.sh",
            "lang.sh",
            "phpinfo.php",
            "install_djvu.sh",
            "plantuml.sh",
            "startRunJobs.sh",
            "upload.ini",
            "update.sh",
        ]:
            self.generate(
                f"{file_name}", f"{self.docker_path}/{file_name}", overwrite=overwrite
            )
        # 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)
        docker.compose.down(volumes=forceRebuild)
        # 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)

        return self.getContainers()

    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 withInitDB:
            if self.config.verbose:
                print("Initializing MediaWiki SQL tables ...")
            dbStatus = self.checkDBConnection()
            if dbStatus.ok:
                # first install extensions
                self.installExtensions()
                # then create and fill database and update it
                self.initDB()
                # then run startUp scripts
                self.startUp()
        if self.config.verbose:
            print(
                f"MediaWiki {self.config.container_base_name} is ready at {self.config.full_url}"
            )

__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.database = "wiki"
    self.dbUser = "wikiuser"
    self.wikiUser = 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)
            )
            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/Special:Version"

            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
    if not docker.compose.is_installed():
        errMsg = """docker compose up needs to be working"""
    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"]}"""
                    )
    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)

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

Source code in mwdocker/docker.py

def createOrModifyWikiUser(self, wikiId, force_overwrite: 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
    """
    wikiUsers = WikiUser.getWikiUsers(lenient=True)
    if wikiId in wikiUsers and not force_overwrite:
        wikiUser = wikiUsers[wikiId]
        if self.config.password != wikiUser.getPassword():
            raise Exception(
                f"wikiUser for wiki {wikiId} already exists but with different password"
            )
        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)
    docker.compose.down(volumes=forceRebuild)
    # switch back to previous current directory
    os.chdir(cwd)

execute(*commands)

execute the given variable list of command strings

Parameters:

Name	Type	Description	Default
commands	str	str - the command strings to be executed ...	()

Source code in mwdocker/docker.py

def execute(self, *commands: str):
    """
    execute the given variable list of command strings

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

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,
            **kwArgs,
        )
        self.optionalWrite(targetPath, content, overwrite)

    except TemplateNotFound:
        print(
            f"no template {templateName} for {self.config.name} {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
    """
    # then generate
    self.generate(
        "mwDockerfile",
        f"{self.docker_path}/Dockerfile",
        composerVersion=self.composerVersion,
        overwrite=overwrite,
    )
    self.generate(
        "mwCompose.yml",
        f"{self.docker_path}/docker-compose.yml",
        mySQLRootPassword=self.config.mySQLRootPassword,
        mySQLPassword=self.config.mySQLPassword,
        container_base_name=self.config.container_base_name,
        overwrite=overwrite,
    )
    self.generate(
        f"mwLocalSettings{self.config.shortVersion}.php",
        f"{self.docker_path}/LocalSettings.php",
        mySQLPassword=self.config.mySQLPassword,
        hostname=self.config.host,
        extensions=self.config.extensionMap.values(),
        mwShortVersion=self.config.shortVersion,
        logo=self.config.logo,
        overwrite=overwrite,
    )
    self.generate(
        f"mwWiki{self.config.shortVersion}.sql",
        f"{self.docker_path}/wiki.sql",
        overwrite=overwrite,
    )
    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
            )
    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 [
        "addCronTabEntry.sh",
        "fixPermissions.sh",
        "initdb.sh",
        "lang.sh",
        "phpinfo.php",
        "install_djvu.sh",
        "plantuml.sh",
        "startRunJobs.sh",
        "upload.ini",
        "update.sh",
    ]:
        self.generate(
            f"{file_name}", f"{self.docker_path}/{file_name}", overwrite=overwrite
        )
    # 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

initDB()

initialize my SQL database

Source code in mwdocker/docker.py

def initDB(self):
    """
    initialize my SQL database
    """
    # restore the mySQL dump data
    self.execute("/root/initdb.sh")
    # update the database e.g. to initialize Semantic MediaWiki tables
    self.execute("/root/update.sh")
    # add an initial sysop user as specified
    self.execute("/root/addSysopUser.sh")

installExtensions()

install all extensions

Source code in mwdocker/docker.py

def installExtensions(self):
    """
    install all extensions
    """
    self.execute("/root/installExtensions.sh")
    self.execute("/root/fixPermissions.sh")

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)

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 withInitDB:
        if self.config.verbose:
            print("Initializing MediaWiki SQL tables ...")
        dbStatus = self.checkDBConnection()
        if dbStatus.ok:
            # first install extensions
            self.installExtensions()
            # then create and fill database and update it
            self.initDB()
            # then run startUp scripts
            self.startUp()
    if self.config.verbose:
        print(
            f"MediaWiki {self.config.container_base_name} is ready at {self.config.full_url}"
        )

startUp()

run startUp scripts

Source code in mwdocker/docker.py

def startUp(self):
    """
    run startUp scripts
    """
    # fix permissions
    self.execute("/root/fixPermissions.sh")
    # add language icons
    self.execute("/root/lang.sh", "--site", "/var/www/html")
    # start cron job
    self.execute("/root/addCronTabEntry.sh")

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)

    return self.getContainers()

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 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

__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)

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

DockerMap

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

Source code in mwdocker/docker.py

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

    @staticmethod
    def getContainerMap():
        """
        get a map/dict of containers by container name
        """
        containerMap = {}
        for container in docker.container.list():
            containerMap[container.name] = container
            pass
        return containerMap

getContainerMap() staticmethod

get a map/dict of containers by container name

Source code in mwdocker/docker.py

@staticmethod
def getContainerMap():
    """
    get a map/dict of containers by container name
    """
    containerMap = {}
    for container in docker.container.list():
        containerMap[container.name] = container
        pass
    return containerMap

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

    @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.getJsonTypeSamples()
        for attr in LOD.getFields(samples):
            if hasattr(self, attr) and self.attr:
                text += f"{delim}{attr}={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 self.attr:
                nameValues += f"|{attr}={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="master"):
        """
        return me as a shell Script command line list

        Args:
            branch(str): the branch to clone
        """
        if self.giturl:
            if "//github.com/wikimedia/" in self.giturl:
                # glone from the branch
                return f"git clone {self.giturl} --single-branch --branch {branch} {self.extension}"
            else:
                return f"git clone {self.giturl} {self.extension}"
        else:
            text = "# no installation script command specified"
            if self.composer:
                text += f"\n# installed with composer require {self.composer}"
            return text

asScript(branch='master')

return me as a shell Script command line list

Parameters:

Name	Type	Description	Default
branch(str)		the branch to clone	required

Source code in mwdocker/mw.py

def asScript(self, branch="master"):
    """
    return me as a shell Script command line list

    Args:
        branch(str): the branch to clone
    """
    if self.giturl:
        if "//github.com/wikimedia/" in self.giturl:
            # glone from the branch
            return f"git clone {self.giturl} --single-branch --branch {branch} {self.extension}"
        else:
            return f"git clone {self.giturl} {self.extension}"
    else:
        text = "# no installation script command specified"
        if self.composer:
            text += f"\n# installed with composer require {self.composer}"
        return text

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 self.attr:
                nameValues += f"|{attr}={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

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

        Args:
            config(MWClusterConfig): the MediaWiki Cluster Configuration to use
        """
        self.config = config
        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
        """
        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.port = self.config.base_port + i
        appConfig.sql_port = self.config.sql_port + i
        # let post_init create a new container_base_name
        if count > 1:
            appConfig.container_base_name = None
        appConfig.__post_init__()
        mwApp = DockerApplication(config=appConfig)
        return mwApp

__init__(config)

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):
    """
    Constructor

    Args:
        config(MWClusterConfig): the MediaWiki Cluster Configuration to use
    """
    self.config = config
    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
    """
    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.port = self.config.base_port + i
    appConfig.sql_port = self.config.sql_port + i
    # let post_init create a new container_base_name
    if count > 1:
        appConfig.container_base_name = 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

main(argv=None)

main program.

Source code in mwdocker/mwcluster.py

def main(argv=None):  # IGNORE:C0111
    """main program."""

    if argv is None:
        argv = sys.argv[1:]

    program_name = "mwcluster"
    program_version = f"v{Version.version}"
    program_build_date = str(Version.updated)
    program_version_message = f"{program_name} ({program_version},{program_build_date})"
    program_license = Version.license

    try:
        # Setup argument parser
        parser = ArgumentParser(
            description=program_license, formatter_class=ArgumentDefaultsHelpFormatter
        )
        mwClusterConfig = MwClusterConfig()
        mwClusterConfig.addArgs(parser)
        parser.add_argument(
            "--about",
            help="show about info [default: %(default)s]",
            action="store_true",
        )
        parser.add_argument(
            "--create", action="store_true", help="create wikis [default: %(default)s]"
        )
        parser.add_argument(
            "--down", action="store_true", help="shutdown wikis [default: %(default)s]"
        )
        parser.add_argument(
            "--check",
            action="store_true",
            help="check the wikis [default: %(default)s]",
        )
        parser.add_argument(
            "--list", action="store_true", help="list the wikis [default: %(default)s]"
        )
        parser.add_argument(
            "-V", "--version", action="version", version=program_version_message
        )
        args = parser.parse_args(argv)
        if args.about:
            print(program_version_message)
            print(f"see {Version.doc_url}")
            webbrowser.open(Version.doc_url)
        else:
            action = None
            withGenerate = False
            if args.check:
                action = "checking docker access"
            elif args.create:
                action = "creating docker compose applications"
                withGenerate = True
            elif args.list:
                action = "listing docker compose wiki applications"
            elif args.down:
                action = "running docker compose down"
            if not action:
                parser.print_usage()
            else:
                print(f"{action} for mediawiki versions {args.versions}")
                # create a MediaWiki Cluster
                mwClusterConfig.fromArgs(args)
                mwCluster = MediaWikiCluster(config=mwClusterConfig)
                mwCluster.createApps(withGenerate=withGenerate)
                if args.check:
                    return mwCluster.check()
                elif args.create:
                    return mwCluster.start(forceRebuild=args.forceRebuild)
                elif args.list:
                    return mwCluster.listWikis()
                elif args.down:
                    return mwCluster.down(forceRebuild=args.forceRebuild)
    except KeyboardInterrupt:
        ### handle keyboard interrupt ###
        return 1
    except Exception as e:
        if DEBUG:
            raise (e)
        indent = len(program_name) * " "
        sys.stderr.write(program_name + ": " + repr(e) + "\n")
        sys.stderr.write(indent + "  for help use --help")
        if args is None:
            print("args could not be parsed")
        elif args.debug:
            print(traceback.format_exc())
        return 2

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 = "2024-08-02"

    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-2024 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)