Skip to content

pyomnigraph API Documentation

basecmd

Created on 31.05.2025

@author: wf

BaseCmd

Base class for Omnigraph-related command line interfaces.

Source code in omnigraph/basecmd.py 
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
class BaseCmd:
    """
    Base class for Omnigraph-related command line interfaces.
    """

    def __init__(self, description: str = None):
        """
        Initialize CLI base.
        """
        self.log = Log()
        self.ogp = OmnigraphPaths()
        self.version = Version()
        self.program_version_message = f"{self.version.name} {self.version.version}"
        if description is None:
            description = self.version.description
        self.parser = None
        self.debug = False
        self.quiet = False
        self.force = False
        self.default_datasets_path = self.ogp.examples_dir / "datasets.yaml"

    def get_arg_parser(self, description: str, version_msg: str) -> ArgumentParser:
        """
        Setup argument parser.

        Args:
            description: CLI description
            version_msg: Version string

        Returns:
            Configured argument parser
        """
        parser = ArgumentParser(description=description, formatter_class=RawDescriptionHelpFormatter)
        parser.add_argument(
            "-a",
            "--about",
            help="show about info [default: %(default)s]",
            action="store_true",
        )
        parser.add_argument(
            "-d",
            "--debug",
            action="store_true",
            help="show debug info [default: %(default)s]",
        )
        parser.add_argument(
            "-ds",
            "--datasets",
            nargs="+",
            default=["wikidata_triplestores"],
            help="datasets to work with - all is an alias for all datasets [default: %(default)s]",
        )
        parser.add_argument(
            "-dc",
            "--datasets-config",
            type=str,
            default=str(self.default_datasets_path),
            help="Path to datasets configuration YAML file [default: %(default)s]",
        )

        parser.add_argument(
            "-f",
            "--force",
            action="store_true",
            help="force actions that would modify existing data [default: %(default)s]",
        )
        rdf_format_choices = [fmt.label for fmt in RdfFormat]

        parser.add_argument(
            "-r",
            "--rdf_format",
            type=str,
            default="turtle",
            choices=rdf_format_choices,
            help="RDF format to use [default: %(default)s]",
        )
        parser.add_argument(
            "-q",
            "--quiet",
            action="store_true",
            help="avoid any output [default: %(default)s]",
        )

        parser.add_argument("-V", "--version", action="version", version=version_msg)
        return parser

    def about(self):
        """
        show about info
        """
        print(self.program_version_message)
        print(f"see {self.version.doc_url}")
        webbrowser.open(self.version.doc_url)

    def handle_args(self, args: Namespace):
        """
        should be extended by specialized subclass.
        """
        self.args = args
        self.debug = args.debug
        self.quiet = args.quiet
        self.force = args.force
        self.datasets = self.getDatasets(yaml_path=args.datasets_config)
        self.rdf_format = RdfFormat.by_label(args.rdf_format)

    def parse_args(self) -> Namespace:
        if not self.parser:
            self.parser = self.get_arg_parser(self.version.description, self.program_version_message)
        args = self.parser.parse_args()
        return args

    def run(self):
        """
        Parse arguments and dispatch to handler.
        """
        args = self.parse_args()
        self.handle_args(args)

    def getDatasets(self, yaml_path: str) -> Dict[str, RdfDataset]:
        """
        Resolve and select datasets to download.

        Args:
            yaml_path: Path to datasets configuration YAML file

        Returns:
            Dict[str, RdfDataset]: selected datasets by name
        """
        datasets = {}
        self.all_datasets = RdfDatasets.ofYaml(yaml_path)
        dataset_names = self.args.datasets
        if "all" in dataset_names:
            dataset_names = list(self.all_datasets.datasets.keys())
        for dataset_name in dataset_names:
            dataset = self.all_datasets.datasets.get(dataset_name)
            if dataset:
                datasets[dataset_name] = dataset
            else:
                self.log.log("⚠️", "omnigraph", f"invalid dataset '{dataset_name}'")
        return datasets

    @classmethod
    def main(cls):
        """
        Entry point for CLI.
        """
        instance = cls()
        args = instance.parse_args()
        instance.handle_args(args)

__init__(description=None)

Initialize CLI base.

Source code in omnigraph/basecmd.py 
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
def __init__(self, description: str = None):
    """
    Initialize CLI base.
    """
    self.log = Log()
    self.ogp = OmnigraphPaths()
    self.version = Version()
    self.program_version_message = f"{self.version.name} {self.version.version}"
    if description is None:
        description = self.version.description
    self.parser = None
    self.debug = False
    self.quiet = False
    self.force = False
    self.default_datasets_path = self.ogp.examples_dir / "datasets.yaml"

about()

show about info

Source code in omnigraph/basecmd.py 
105
106
107
108
109
110
111
def about(self):
    """
    show about info
    """
    print(self.program_version_message)
    print(f"see {self.version.doc_url}")
    webbrowser.open(self.version.doc_url)

getDatasets(yaml_path)

Resolve and select datasets to download.

Parameters:

Name Type Description Default
yaml_path str

Path to datasets configuration YAML file

 required

Returns:

Type Description
Dict[str, RdfDataset]

Dict[str, RdfDataset]: selected datasets by name
Source code in omnigraph/basecmd.py 
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
def getDatasets(self, yaml_path: str) -> Dict[str, RdfDataset]:
    """
    Resolve and select datasets to download.

    Args:
        yaml_path: Path to datasets configuration YAML file

    Returns:
        Dict[str, RdfDataset]: selected datasets by name
    """
    datasets = {}
    self.all_datasets = RdfDatasets.ofYaml(yaml_path)
    dataset_names = self.args.datasets
    if "all" in dataset_names:
        dataset_names = list(self.all_datasets.datasets.keys())
    for dataset_name in dataset_names:
        dataset = self.all_datasets.datasets.get(dataset_name)
        if dataset:
            datasets[dataset_name] = dataset
        else:
            self.log.log("⚠️", "omnigraph", f"invalid dataset '{dataset_name}'")
    return datasets

get_arg_parser(description, version_msg)

Setup argument parser.

Parameters:

Name Type Description Default
description str

CLI description

 required
version_msg str

Version string

 required

Returns:

Type Description
ArgumentParser

Configured argument parser
Source code in omnigraph/basecmd.py 
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
def get_arg_parser(self, description: str, version_msg: str) -> ArgumentParser:
    """
    Setup argument parser.

    Args:
        description: CLI description
        version_msg: Version string

    Returns:
        Configured argument parser
    """
    parser = ArgumentParser(description=description, formatter_class=RawDescriptionHelpFormatter)
    parser.add_argument(
        "-a",
        "--about",
        help="show about info [default: %(default)s]",
        action="store_true",
    )
    parser.add_argument(
        "-d",
        "--debug",
        action="store_true",
        help="show debug info [default: %(default)s]",
    )
    parser.add_argument(
        "-ds",
        "--datasets",
        nargs="+",
        default=["wikidata_triplestores"],
        help="datasets to work with - all is an alias for all datasets [default: %(default)s]",
    )
    parser.add_argument(
        "-dc",
        "--datasets-config",
        type=str,
        default=str(self.default_datasets_path),
        help="Path to datasets configuration YAML file [default: %(default)s]",
    )

    parser.add_argument(
        "-f",
        "--force",
        action="store_true",
        help="force actions that would modify existing data [default: %(default)s]",
    )
    rdf_format_choices = [fmt.label for fmt in RdfFormat]

    parser.add_argument(
        "-r",
        "--rdf_format",
        type=str,
        default="turtle",
        choices=rdf_format_choices,
        help="RDF format to use [default: %(default)s]",
    )
    parser.add_argument(
        "-q",
        "--quiet",
        action="store_true",
        help="avoid any output [default: %(default)s]",
    )

    parser.add_argument("-V", "--version", action="version", version=version_msg)
    return parser

handle_args(args)

should be extended by specialized subclass.

Source code in omnigraph/basecmd.py 
113
114
115
116
117
118
119
120
121
122
def handle_args(self, args: Namespace):
    """
    should be extended by specialized subclass.
    """
    self.args = args
    self.debug = args.debug
    self.quiet = args.quiet
    self.force = args.force
    self.datasets = self.getDatasets(yaml_path=args.datasets_config)
    self.rdf_format = RdfFormat.by_label(args.rdf_format)

main() classmethod

Entry point for CLI.

Source code in omnigraph/basecmd.py 
160
161
162
163
164
165
166
167
@classmethod
def main(cls):
    """
    Entry point for CLI.
    """
    instance = cls()
    args = instance.parse_args()
    instance.handle_args(args)

run()

Parse arguments and dispatch to handler.

Source code in omnigraph/basecmd.py 
130
131
132
133
134
135
def run(self):
    """
    Parse arguments and dispatch to handler.
    """
    args = self.parse_args()
    self.handle_args(args)

ominigraph_paths

Created on 2025-05-27

@author: wf

OmnigraphPaths

Omnigraph Default Paths

Source code in omnigraph/ominigraph_paths.py 
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
class OmnigraphPaths:
    """
    Omnigraph Default Paths
    """

    def __init__(self, home_dir: Optional[Path] = None):
        """
        Initialize default Omnigraph paths

        Args:
            home_dir: Optional custom home directory path (default: Path.home())
        """
        self.home_dir = home_dir if home_dir else Path.home()
        self.omnigraph_dir = self.home_dir / ".omnigraph"
        self.omnigraph_dir.mkdir(parents=True, exist_ok=True)
        self.dumps_dir = self.omnigraph_dir / "rdf_dumps"
        self.dumps_dir.mkdir(exist_ok=True)
        self.examples_dir = (Path(__file__).parent / "resources" / "examples").resolve()

__init__(home_dir=None)

Initialize default Omnigraph paths

Parameters:

Name Type Description Default
home_dir Optional[Path]

Optional custom home directory path (default: Path.home())

 None
Source code in omnigraph/ominigraph_paths.py 
16
17
18
19
20
21
22
23
24
25
26
27
28
def __init__(self, home_dir: Optional[Path] = None):
    """
    Initialize default Omnigraph paths

    Args:
        home_dir: Optional custom home directory path (default: Path.home())
    """
    self.home_dir = home_dir if home_dir else Path.home()
    self.omnigraph_dir = self.home_dir / ".omnigraph"
    self.omnigraph_dir.mkdir(parents=True, exist_ok=True)
    self.dumps_dir = self.omnigraph_dir / "rdf_dumps"
    self.dumps_dir.mkdir(exist_ok=True)
    self.examples_dir = (Path(__file__).parent / "resources" / "examples").resolve()

omnigraph_cmd

Created on 2025-05-28

@author: wf

OmnigraphCmd

Bases: BaseCmd

Command line interface for omnigraph.

Source code in omnigraph/omnigraph_cmd.py 
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
class OmnigraphCmd(BaseCmd):
    """
    Command line interface for omnigraph.
    """

    def __init__(self):
        """
        Initialize command line interface.
        """
        self.ogp = OmnigraphPaths()
        self.default_yaml_path = self.ogp.examples_dir / "servers.yaml"
        self.env = ServerEnv()
        self.omni_server = OmniServer(env=self.env)
        self.server_cmds = self.omni_server.get_server_commands()
        self.available_cmds = ", ".join(self.server_cmds.keys())
        self.prefixes_yaml_path = self.ogp.examples_dir / "prefixes.yaml"
        self.prefix_configs = PrefixConfigs.of_yaml(self.prefixes_yaml_path)

        super().__init__(description="Manage SPARQL server configurations and command execution")

    def get_arg_parser(self, description: str, version_msg: str) -> ArgumentParser:
        """
        Extend base parser with Omnigraph-specific arguments.
        """
        parser = super().get_arg_parser(description, version_msg)
        parser.add_argument(
            "--apache",
            help="create apache configuration file for the given server(s)",
        )
        parser.add_argument(
            "-c",
            "--config",
            type=str,
            default=str(self.default_yaml_path),
            help="Path to server configuration YAML file [default: %(default)s]",
        )
        parser.add_argument("--cmd", nargs="+", help=f"commands to execute on servers: {self.available_cmds}")
        parser.add_argument(
            "-df", "--doc-format", default="plain", help="The document format to use [default: %(default)s]"
        )
        parser.add_argument(
            "-gepy",
            "--endpoints-yaml",
            help="Generate and endpoints yaml file for the active servers [default: %(default)s]",
        )
        parser.add_argument(
            "-ii",
            "--include-inactive",
            action="store_true",
            help="Include inactive servers in available server list[default: %(default)s]",
        )
        parser.add_argument(
            "-l", "--list-servers", action="store_true", help="List available servers [default: %(default)s]"
        )
        parser.add_argument(
            "--test",
            action="store_true",
            help="use test environment [default: %(default)s]",
        )
        parser.add_argument(
            "-s",
            "--servers",
            nargs="+",
            default=["blazegraph"],
            help="servers to work with - 'all' selects all configured servers [default: %(default)s]",
        )
        parser.add_argument(
            "-v",
            "--verbose",
            action="store_true",
            help="show verbose output [default: %(default)s]",
        )
        return parser

    def getServers(self) -> Dict[str, SparqlServer]:
        """
        Get the active servers from configuration.
        """
        servers = {}
        server_names = self.args.servers
        if "all" in server_names:
            server_names = list(self.all_servers.keys())
        for server_name in server_names:
            server = self.all_servers.get(server_name)
            if server:
                server.config.base_data_dir = self.ogp.omnigraph_dir / server.name / "data"
                server.config.data_dir = server.config.base_data_dir / server.config.dataset
                server.config.data_dir.mkdir(parents=True, exist_ok=True)
                if server.config.dumps_dir is None:
                    self.configure_dumps_dir(server)
                server.config.rdf_format = self.args.rdf_format
                servers[server_name] = server
            else:
                self.log.log("⚠️", "omnigraph", f"invalid or inactive server '{server_name}'")
        return servers

    def configure_dumps_dir(self, server: SparqlServer, dataset: RdfDataset = None):
        """
        Configure dumps directory for a server based on dataset.

        Args:
            server: Server instance to configure
            dataset: RdfDataset instance
        """
        if dataset is None:
            server.config.dumps_dir = self.ogp.examples_dir
        elif dataset.rdf_file:
            # If rdf_file is specified, set dumps_dir to the parent directory of the file
            rdf_file_path = Path(dataset.rdf_file).expanduser().resolve()
            server.config.dumps_dir = rdf_file_path.parent
        else:
            server.config.dumps_dir = self.ogp.dumps_dir / dataset.id

    def run_single_cmd(self, server: SparqlServer, cmd: str) -> bool:
        """
        Run a single command on a server.

        Args:
            server: Server instance
            cmd: Command name to run

        Returns:
            bool: True if command was successfully run
        """
        s_cmd_factory = self.server_cmds.get(cmd)
        s_cmd = s_cmd_factory(server) if s_cmd_factory else None
        if s_cmd:
            s_cmd.run(verbose=not self.quiet)
            return True
        else:
            print(f"unsupported command {cmd}")
            return False

    def load_iterator(self, server):
        """
        Iterator for load command that configures dumps_dir for each dataset.
        """
        total_datasets = len(self.datasets)
        for i, (dataset_name, dataset) in enumerate(self.datasets.items(), 1):
            if not self.quiet:
                print(f"loading {dataset_name} ({i}/{total_datasets})...")
            self.configure_dumps_dir(server, dataset)
            yield

        if not self.quiet:
            print(f"Loaded {total_datasets} dataset(s)")

    def run_cmds(self, server: SparqlServer, cmds: List[str]) -> bool:
        """
        Run commands on a specific server.
        """
        handled = False
        if cmds:
            for cmd in cmds:
                # issue #38: upload needs the per-dataset dumps_dir just like load
                if cmd in ("load", "upload"):
                    cmd_iterator = self.load_iterator(server)
                else:
                    cmd_iterator = iter([None])  # Single iteration

                for _ in cmd_iterator:
                    if self.run_single_cmd(server, cmd):
                        handled = True
        return handled

    def handle_args(self, args: Namespace):
        """
        Handle parsed CLI arguments.

        Args:
            args: parsed argument namespace
        """
        super().handle_args(args)
        self.all_servers = {}
        if Path(self.args.config).exists():
            env = ServerEnv(force=self.force, debug=self.debug, verbose=self.args.verbose)
            patch_config = None
            if self.args.test:
                patch_config = lambda config: OmniServer.patch_test_config(config, self.ogp)
            omni_server = OmniServer(env=env, patch_config=patch_config)
            self.all_servers = omni_server.servers(self.args.config, filter_active=not self.args.include_inactive)
        else:
            print(f"Config file not found: {self.args.config}")
        self.servers = self.getServers()

        if self.args.about:
            self.about()
            print(f"{len(self.all_servers)} servers configured - {len(self.servers)} active")
            for _name, server in self.servers.items():
                print(f"  {server.full_name}")

        if self.args.endpoints_yaml:
            output_path = self.args.endpoints_yaml
            _yaml_content = self.omni_server.generate_endpoints_yaml(
                self.servers, self.prefix_configs, output_path=output_path
            )
            pass
        if self.args.apache:
            if self.args.apache:
                for server in self.servers.values():
                    config = server.config
                    print(config.to_apache_config(version=self.version, domain=self.args.apache))

        if self.args.list_servers:
            table_format = self.args.doc_format if self.args.doc_format != "plain" else "simple"
            markup = self.omni_server.list_servers(self.all_servers, table_format)
            print(markup)

        cmds = list(self.args.cmd or [])
        if len(cmds) > 0:
            for server in self.servers.values():
                if not self.quiet:
                    print(f"{server.flag}  {server.full_name}:")
                try:
                    self.run_cmds(server, cmds=cmds)
                except Exception as ex:
                    server.handle_exception(str(self.args.cmd), ex)

__init__()

Initialize command line interface.

Source code in omnigraph/omnigraph_cmd.py 
25
26
27
28
29
30
31
32
33
34
35
36
37
38
def __init__(self):
    """
    Initialize command line interface.
    """
    self.ogp = OmnigraphPaths()
    self.default_yaml_path = self.ogp.examples_dir / "servers.yaml"
    self.env = ServerEnv()
    self.omni_server = OmniServer(env=self.env)
    self.server_cmds = self.omni_server.get_server_commands()
    self.available_cmds = ", ".join(self.server_cmds.keys())
    self.prefixes_yaml_path = self.ogp.examples_dir / "prefixes.yaml"
    self.prefix_configs = PrefixConfigs.of_yaml(self.prefixes_yaml_path)

    super().__init__(description="Manage SPARQL server configurations and command execution")

configure_dumps_dir(server, dataset=None)

Configure dumps directory for a server based on dataset.

Parameters:

Name Type Description Default
server SparqlServer

Server instance to configure

 required
dataset RdfDataset

RdfDataset instance

 None
Source code in omnigraph/omnigraph_cmd.py 
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
def configure_dumps_dir(self, server: SparqlServer, dataset: RdfDataset = None):
    """
    Configure dumps directory for a server based on dataset.

    Args:
        server: Server instance to configure
        dataset: RdfDataset instance
    """
    if dataset is None:
        server.config.dumps_dir = self.ogp.examples_dir
    elif dataset.rdf_file:
        # If rdf_file is specified, set dumps_dir to the parent directory of the file
        rdf_file_path = Path(dataset.rdf_file).expanduser().resolve()
        server.config.dumps_dir = rdf_file_path.parent
    else:
        server.config.dumps_dir = self.ogp.dumps_dir / dataset.id

getServers()

Get the active servers from configuration.

Source code in omnigraph/omnigraph_cmd.py 
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
def getServers(self) -> Dict[str, SparqlServer]:
    """
    Get the active servers from configuration.
    """
    servers = {}
    server_names = self.args.servers
    if "all" in server_names:
        server_names = list(self.all_servers.keys())
    for server_name in server_names:
        server = self.all_servers.get(server_name)
        if server:
            server.config.base_data_dir = self.ogp.omnigraph_dir / server.name / "data"
            server.config.data_dir = server.config.base_data_dir / server.config.dataset
            server.config.data_dir.mkdir(parents=True, exist_ok=True)
            if server.config.dumps_dir is None:
                self.configure_dumps_dir(server)
            server.config.rdf_format = self.args.rdf_format
            servers[server_name] = server
        else:
            self.log.log("⚠️", "omnigraph", f"invalid or inactive server '{server_name}'")
    return servers

get_arg_parser(description, version_msg)

Extend base parser with Omnigraph-specific arguments.

Source code in omnigraph/omnigraph_cmd.py 
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
def get_arg_parser(self, description: str, version_msg: str) -> ArgumentParser:
    """
    Extend base parser with Omnigraph-specific arguments.
    """
    parser = super().get_arg_parser(description, version_msg)
    parser.add_argument(
        "--apache",
        help="create apache configuration file for the given server(s)",
    )
    parser.add_argument(
        "-c",
        "--config",
        type=str,
        default=str(self.default_yaml_path),
        help="Path to server configuration YAML file [default: %(default)s]",
    )
    parser.add_argument("--cmd", nargs="+", help=f"commands to execute on servers: {self.available_cmds}")
    parser.add_argument(
        "-df", "--doc-format", default="plain", help="The document format to use [default: %(default)s]"
    )
    parser.add_argument(
        "-gepy",
        "--endpoints-yaml",
        help="Generate and endpoints yaml file for the active servers [default: %(default)s]",
    )
    parser.add_argument(
        "-ii",
        "--include-inactive",
        action="store_true",
        help="Include inactive servers in available server list[default: %(default)s]",
    )
    parser.add_argument(
        "-l", "--list-servers", action="store_true", help="List available servers [default: %(default)s]"
    )
    parser.add_argument(
        "--test",
        action="store_true",
        help="use test environment [default: %(default)s]",
    )
    parser.add_argument(
        "-s",
        "--servers",
        nargs="+",
        default=["blazegraph"],
        help="servers to work with - 'all' selects all configured servers [default: %(default)s]",
    )
    parser.add_argument(
        "-v",
        "--verbose",
        action="store_true",
        help="show verbose output [default: %(default)s]",
    )
    return parser

handle_args(args)

Handle parsed CLI arguments.

Parameters:

Name Type Description Default
args Namespace

parsed argument namespace

 required
Source code in omnigraph/omnigraph_cmd.py 
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
def handle_args(self, args: Namespace):
    """
    Handle parsed CLI arguments.

    Args:
        args: parsed argument namespace
    """
    super().handle_args(args)
    self.all_servers = {}
    if Path(self.args.config).exists():
        env = ServerEnv(force=self.force, debug=self.debug, verbose=self.args.verbose)
        patch_config = None
        if self.args.test:
            patch_config = lambda config: OmniServer.patch_test_config(config, self.ogp)
        omni_server = OmniServer(env=env, patch_config=patch_config)
        self.all_servers = omni_server.servers(self.args.config, filter_active=not self.args.include_inactive)
    else:
        print(f"Config file not found: {self.args.config}")
    self.servers = self.getServers()

    if self.args.about:
        self.about()
        print(f"{len(self.all_servers)} servers configured - {len(self.servers)} active")
        for _name, server in self.servers.items():
            print(f"  {server.full_name}")

    if self.args.endpoints_yaml:
        output_path = self.args.endpoints_yaml
        _yaml_content = self.omni_server.generate_endpoints_yaml(
            self.servers, self.prefix_configs, output_path=output_path
        )
        pass
    if self.args.apache:
        if self.args.apache:
            for server in self.servers.values():
                config = server.config
                print(config.to_apache_config(version=self.version, domain=self.args.apache))

    if self.args.list_servers:
        table_format = self.args.doc_format if self.args.doc_format != "plain" else "simple"
        markup = self.omni_server.list_servers(self.all_servers, table_format)
        print(markup)

    cmds = list(self.args.cmd or [])
    if len(cmds) > 0:
        for server in self.servers.values():
            if not self.quiet:
                print(f"{server.flag}  {server.full_name}:")
            try:
                self.run_cmds(server, cmds=cmds)
            except Exception as ex:
                server.handle_exception(str(self.args.cmd), ex)

load_iterator(server)

Iterator for load command that configures dumps_dir for each dataset.

Source code in omnigraph/omnigraph_cmd.py 
153
154
155
156
157
158
159
160
161
162
163
164
165
def load_iterator(self, server):
    """
    Iterator for load command that configures dumps_dir for each dataset.
    """
    total_datasets = len(self.datasets)
    for i, (dataset_name, dataset) in enumerate(self.datasets.items(), 1):
        if not self.quiet:
            print(f"loading {dataset_name} ({i}/{total_datasets})...")
        self.configure_dumps_dir(server, dataset)
        yield

    if not self.quiet:
        print(f"Loaded {total_datasets} dataset(s)")

run_cmds(server, cmds)

Run commands on a specific server.

Source code in omnigraph/omnigraph_cmd.py 
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
def run_cmds(self, server: SparqlServer, cmds: List[str]) -> bool:
    """
    Run commands on a specific server.
    """
    handled = False
    if cmds:
        for cmd in cmds:
            # issue #38: upload needs the per-dataset dumps_dir just like load
            if cmd in ("load", "upload"):
                cmd_iterator = self.load_iterator(server)
            else:
                cmd_iterator = iter([None])  # Single iteration

            for _ in cmd_iterator:
                if self.run_single_cmd(server, cmd):
                    handled = True
    return handled

run_single_cmd(server, cmd)

Run a single command on a server.

Parameters:

Name Type Description Default
server SparqlServer

Server instance

 required
cmd str

Command name to run

 required

Returns:

Name Type Description
bool bool

True if command was successfully run
Source code in omnigraph/omnigraph_cmd.py 
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
def run_single_cmd(self, server: SparqlServer, cmd: str) -> bool:
    """
    Run a single command on a server.

    Args:
        server: Server instance
        cmd: Command name to run

    Returns:
        bool: True if command was successfully run
    """
    s_cmd_factory = self.server_cmds.get(cmd)
    s_cmd = s_cmd_factory(server) if s_cmd_factory else None
    if s_cmd:
        s_cmd.run(verbose=not self.quiet)
        return True
    else:
        print(f"unsupported command {cmd}")
        return False

omniserver

Created on 2025-05-28

@author: wf

OmniServer

Factory class for creating and managing SPARQL server instances.

Source code in omnigraph/omniserver.py 
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
class OmniServer:
    """
    Factory class for creating and managing SPARQL server instances.
    """

    def __init__(self, env: ServerEnv, patch_config: Callable = None):
        """
        constructor
        """
        self.env = env
        self.patch_config = patch_config

    @staticmethod
    def patch_test_config(config: ServerConfig, ogp: OmnigraphPaths):
        config.base_data_dir = ogp.omnigraph_dir / "test" / config.name / "data"
        config.data_dir = config.base_data_dir / config.dataset
        config.data_dir.mkdir(parents=True, exist_ok=True)
        config.container_name = f"{config.container_name}-test"
        config.port = config.test_port
        # make sure the port is reconfigured for test
        config.base_url = None
        pass

    def get_server_commands(self) -> Dict[str, Callable[[SparqlServer], ServerCmd]]:
        """
        Get available server commands as factory functions.

        Returns:
            Dictionary mapping command names to ServerCmd factories
        """

        def title(action, s):
            return f"{action} {s.name} ({s.config.container_name})"

        server_cmds = {
            "bash": lambda s: ServerCmd(title("bash into", s), s.bash),
            "clear": lambda s: ServerCmd(title("clear", s), s.clear),
            "count": lambda s: ServerCmd(title("triple count", s), s.count_triples),
            "info": lambda s: ServerCmd(title("info", s), s.docker_info),
            "load": lambda s: ServerCmd(title("load dumps", s), s.load_dump_files),
            "logs": lambda s: ServerCmd(title("logs of", s), s.logs),
            "needed": lambda s: ServerCmd(title("check needed software for", s), s.check_needed_software),
            "rm": lambda s: ServerCmd(title("remove", s), s.rm),
            "start": lambda s: ServerCmd(title("start", s), s.start),
            "status": lambda s: ServerCmd(title("status", s), s.status_info),
            "stop": lambda s: ServerCmd(title("stop", s), s.stop),
            "upload": lambda s: ServerCmd(title("native bulk upload dumps", s), s.upload_dump_files),
            "webui": lambda s: ServerCmd(title("webui", s), s.webui),
        }
        return server_cmds

    def server4Config(self, config: ServerConfig) -> SparqlServer:
        """
        Create a SparqlServer instance based on server type in config.

        Args:
            config: ServerConfig with server type and settings

        Returns:
            SparqlServer instance of appropriate type
        """
        if self.patch_config:
            self.patch_config(config)
        config_dict = asdict(config)

        server_mappings = {
            "blazegraph": (BlazegraphConfig, Blazegraph),
            "graphdb": (GraphDBConfig, GraphDB),
            "jena": (JenaConfig, Jena),
            "millenniumdb": (MillenniumDBConfig, MillenniumDB),
            "oxigraph": (OxigraphConfig, Oxigraph),
            "qlever": (QLeverConfig, QLever),
            "stardog": (StardogConfig, Stardog),
            "virtuoso": (VirtuosoConfig, Virtuoso),
        }

        if config.server not in server_mappings:
            raise ValueError(f"Knowledge Graph Server {config.server} not supported yet")

        config_class, server_class = server_mappings[config.server]
        server_config = config_class(**config_dict)
        server_instance = server_class(config=server_config, env=self.env)

        return server_instance

    def servers(self, yaml_path: Path, filter_active: bool = True) -> Dict[str, SparqlServer]:
        """
        Load active servers from YAML configuration.

        Args:
            yaml_path: Path to YAML configuration file
            filter_active: if true filter active servers

        Returns:
            Dictionary mapping server names to SparqlServer instances
        """
        server_configs = ServerConfigs.ofYaml(yaml_path)
        servers_dict = {}

        for server_name, config in server_configs.servers.items():
            if config.active or not filter_active:
                server_instance = self.server4Config(config)
                if server_instance:
                    servers_dict[server_name] = server_instance

        return servers_dict

    def list_servers(self, servers: Dict[str, SparqlServer], table_format: str, host: str = "localhost") -> str:
        """
        Generate formatted table of servers.

        Args:
            servers: Dictionary of server instances
            table_format: Table format for tabulate
            host: Host for server links (default: localhost)

        Returns:
            str: Formatted table markup
        """
        headers = ["Active", "Name", "Container Name", "Wikidata", "Image", "Port", "Test Port", "Dataset", "User"]
        table_data = []

        def format_link(text: str, url: str, format_type: str) -> str:
            """Format link based on table format."""
            if format_type == "plain" or format_type == "simple":
                return text
            elif format_type == "html":
                return f'<a href="{url}">{text}</a>'
            elif format_type == "mediawiki":
                return f"[{url} {text}]"
            elif format_type == "rst":
                return f"`{text} <{url}>`_"
            elif format_type == "github":
                return f"[{text}]({url})"
            else:
                return text

        for server in servers.values():
            active_str = server.flag

            wikidata_id = getattr(server.config, "wikidata_id", "")
            wikidata_link = (
                format_link(wikidata_id, f"https://www.wikidata.org/wiki/{wikidata_id}", table_format)
                if wikidata_id
                else ""
            )

            server_host = getattr(server.config, "host", host)
            if server_host == "localhost" and host != "localhost":
                server_host = host
            server_port = getattr(server.config, "port", "")
            server_url = f"http://{server_host}:{server_port}" if server_port else ""
            server_name_link = format_link(server.name, server_url, table_format) if server_url else server.name
            image = getattr(server.config, "image", "")
            image_link = image
            if image:
                # Remove tag (everything after :) and create Docker Hub link
                image_name = image.split(":")[0]
                docker_url = f"https://hub.docker.com/r/{image_name}"
                image_link = format_link(image, docker_url, table_format)

            table_data.append(
                [
                    active_str,
                    server_name_link,
                    server.config.container_name,
                    wikidata_link,
                    image_link,
                    server_port,
                    getattr(server.config, "test_port", ""),
                    getattr(server.config, "dataset", ""),
                    getattr(server.config, "auth_user", ""),
                ]
            )

        markup = tabulate(table_data, headers=headers, tablefmt=table_format)
        return markup

    def generate_endpoints_yaml(
        self, servers: Dict[str, SparqlServer], prefix_configs: PrefixConfigs, output_path: str = None
    ) -> str:
        """
        Generate endpoints.yaml from server configurations.
        """
        yaml_entries = []

        for server in servers.values():
            if server.config.active:
                prefix_sets = getattr(server.config, "prefix_sets", ["rdf"])
                prefixes_text = prefix_configs.get_selected_declarations(prefix_sets)
                # optional elements
                auth = "\n  auth: BASIC" if server.config.auth_user else ""
                user = f"\n  user: {server.config.auth_user}" if server.config.auth_user else ""
                passwd = f"\n  passwd: {server.config.auth_password}" if server.config.auth_password else ""
                # Indent prefixes for literal block scalar (4 spaces)
                indented_prefixes = "\n".join(f"    {line}" for line in prefixes_text.split("\n") if line.strip())

                entry = f"""{server.name}:
  method: {getattr(server.config, 'method', 'POST')}
  lang: sparql
  name: {server.name}
  endpoint: {server.config.sparql_url}
  website: {server.config.base_url}
  database: {server.config.server}{auth}{user}{passwd}
  prefixes: |
{indented_prefixes}"""

                yaml_entries.append(entry)

        yaml_header = "# SPARQL endpoints for snapquery, sparqlquery and omnigraph tools\n"
        yaml_header += server.config.generator_header() + "\n"
        yaml_content = yaml_header + "\n".join(yaml_entries)

        if output_path:
            with open(output_path, "w") as f:
                f.write(yaml_content)

        return yaml_content

__init__(env, patch_config=None)

constructor

Source code in omnigraph/omniserver.py 
32
33
34
35
36
37
def __init__(self, env: ServerEnv, patch_config: Callable = None):
    """
    constructor
    """
    self.env = env
    self.patch_config = patch_config

generate_endpoints_yaml(servers, prefix_configs, output_path=None)

Generate endpoints.yaml from server configurations.

Source code in omnigraph/omniserver.py 
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
    def generate_endpoints_yaml(
        self, servers: Dict[str, SparqlServer], prefix_configs: PrefixConfigs, output_path: str = None
    ) -> str:
        """
        Generate endpoints.yaml from server configurations.
        """
        yaml_entries = []

        for server in servers.values():
            if server.config.active:
                prefix_sets = getattr(server.config, "prefix_sets", ["rdf"])
                prefixes_text = prefix_configs.get_selected_declarations(prefix_sets)
                # optional elements
                auth = "\n  auth: BASIC" if server.config.auth_user else ""
                user = f"\n  user: {server.config.auth_user}" if server.config.auth_user else ""
                passwd = f"\n  passwd: {server.config.auth_password}" if server.config.auth_password else ""
                # Indent prefixes for literal block scalar (4 spaces)
                indented_prefixes = "\n".join(f"    {line}" for line in prefixes_text.split("\n") if line.strip())

                entry = f"""{server.name}:
  method: {getattr(server.config, 'method', 'POST')}
  lang: sparql
  name: {server.name}
  endpoint: {server.config.sparql_url}
  website: {server.config.base_url}
  database: {server.config.server}{auth}{user}{passwd}
  prefixes: |
{indented_prefixes}"""

                yaml_entries.append(entry)

        yaml_header = "# SPARQL endpoints for snapquery, sparqlquery and omnigraph tools\n"
        yaml_header += server.config.generator_header() + "\n"
        yaml_content = yaml_header + "\n".join(yaml_entries)

        if output_path:
            with open(output_path, "w") as f:
                f.write(yaml_content)

        return yaml_content

get_server_commands()

Get available server commands as factory functions.

Returns:

Type Description
Dict[str, Callable[[SparqlServer], ServerCmd]]

Dictionary mapping command names to ServerCmd factories
Source code in omnigraph/omniserver.py 
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
def get_server_commands(self) -> Dict[str, Callable[[SparqlServer], ServerCmd]]:
    """
    Get available server commands as factory functions.

    Returns:
        Dictionary mapping command names to ServerCmd factories
    """

    def title(action, s):
        return f"{action} {s.name} ({s.config.container_name})"

    server_cmds = {
        "bash": lambda s: ServerCmd(title("bash into", s), s.bash),
        "clear": lambda s: ServerCmd(title("clear", s), s.clear),
        "count": lambda s: ServerCmd(title("triple count", s), s.count_triples),
        "info": lambda s: ServerCmd(title("info", s), s.docker_info),
        "load": lambda s: ServerCmd(title("load dumps", s), s.load_dump_files),
        "logs": lambda s: ServerCmd(title("logs of", s), s.logs),
        "needed": lambda s: ServerCmd(title("check needed software for", s), s.check_needed_software),
        "rm": lambda s: ServerCmd(title("remove", s), s.rm),
        "start": lambda s: ServerCmd(title("start", s), s.start),
        "status": lambda s: ServerCmd(title("status", s), s.status_info),
        "stop": lambda s: ServerCmd(title("stop", s), s.stop),
        "upload": lambda s: ServerCmd(title("native bulk upload dumps", s), s.upload_dump_files),
        "webui": lambda s: ServerCmd(title("webui", s), s.webui),
    }
    return server_cmds

list_servers(servers, table_format, host='localhost')

Generate formatted table of servers.

Parameters:

Name Type Description Default
servers Dict[str, SparqlServer]

Dictionary of server instances

 required
table_format str

Table format for tabulate

 required
host str

Host for server links (default: localhost)

 'localhost'

Returns:

Name Type Description
str str

Formatted table markup
Source code in omnigraph/omniserver.py 
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
def list_servers(self, servers: Dict[str, SparqlServer], table_format: str, host: str = "localhost") -> str:
    """
    Generate formatted table of servers.

    Args:
        servers: Dictionary of server instances
        table_format: Table format for tabulate
        host: Host for server links (default: localhost)

    Returns:
        str: Formatted table markup
    """
    headers = ["Active", "Name", "Container Name", "Wikidata", "Image", "Port", "Test Port", "Dataset", "User"]
    table_data = []

    def format_link(text: str, url: str, format_type: str) -> str:
        """Format link based on table format."""
        if format_type == "plain" or format_type == "simple":
            return text
        elif format_type == "html":
            return f'<a href="{url}">{text}</a>'
        elif format_type == "mediawiki":
            return f"[{url} {text}]"
        elif format_type == "rst":
            return f"`{text} <{url}>`_"
        elif format_type == "github":
            return f"[{text}]({url})"
        else:
            return text

    for server in servers.values():
        active_str = server.flag

        wikidata_id = getattr(server.config, "wikidata_id", "")
        wikidata_link = (
            format_link(wikidata_id, f"https://www.wikidata.org/wiki/{wikidata_id}", table_format)
            if wikidata_id
            else ""
        )

        server_host = getattr(server.config, "host", host)
        if server_host == "localhost" and host != "localhost":
            server_host = host
        server_port = getattr(server.config, "port", "")
        server_url = f"http://{server_host}:{server_port}" if server_port else ""
        server_name_link = format_link(server.name, server_url, table_format) if server_url else server.name
        image = getattr(server.config, "image", "")
        image_link = image
        if image:
            # Remove tag (everything after :) and create Docker Hub link
            image_name = image.split(":")[0]
            docker_url = f"https://hub.docker.com/r/{image_name}"
            image_link = format_link(image, docker_url, table_format)

        table_data.append(
            [
                active_str,
                server_name_link,
                server.config.container_name,
                wikidata_link,
                image_link,
                server_port,
                getattr(server.config, "test_port", ""),
                getattr(server.config, "dataset", ""),
                getattr(server.config, "auth_user", ""),
            ]
        )

    markup = tabulate(table_data, headers=headers, tablefmt=table_format)
    return markup

server4Config(config)

Create a SparqlServer instance based on server type in config.

Parameters:

Name Type Description Default
config ServerConfig

ServerConfig with server type and settings

 required

Returns:

Type Description
SparqlServer

SparqlServer instance of appropriate type
Source code in omnigraph/omniserver.py 
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
def server4Config(self, config: ServerConfig) -> SparqlServer:
    """
    Create a SparqlServer instance based on server type in config.

    Args:
        config: ServerConfig with server type and settings

    Returns:
        SparqlServer instance of appropriate type
    """
    if self.patch_config:
        self.patch_config(config)
    config_dict = asdict(config)

    server_mappings = {
        "blazegraph": (BlazegraphConfig, Blazegraph),
        "graphdb": (GraphDBConfig, GraphDB),
        "jena": (JenaConfig, Jena),
        "millenniumdb": (MillenniumDBConfig, MillenniumDB),
        "oxigraph": (OxigraphConfig, Oxigraph),
        "qlever": (QLeverConfig, QLever),
        "stardog": (StardogConfig, Stardog),
        "virtuoso": (VirtuosoConfig, Virtuoso),
    }

    if config.server not in server_mappings:
        raise ValueError(f"Knowledge Graph Server {config.server} not supported yet")

    config_class, server_class = server_mappings[config.server]
    server_config = config_class(**config_dict)
    server_instance = server_class(config=server_config, env=self.env)

    return server_instance

servers(yaml_path, filter_active=True)

Load active servers from YAML configuration.

Parameters:

Name Type Description Default
yaml_path Path

Path to YAML configuration file

 required
filter_active bool

if true filter active servers

 True

Returns:

Type Description
Dict[str, SparqlServer]

Dictionary mapping server names to SparqlServer instances
Source code in omnigraph/omniserver.py 
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
def servers(self, yaml_path: Path, filter_active: bool = True) -> Dict[str, SparqlServer]:
    """
    Load active servers from YAML configuration.

    Args:
        yaml_path: Path to YAML configuration file
        filter_active: if true filter active servers

    Returns:
        Dictionary mapping server names to SparqlServer instances
    """
    server_configs = ServerConfigs.ofYaml(yaml_path)
    servers_dict = {}

    for server_name, config in server_configs.servers.items():
        if config.active or not filter_active:
            server_instance = self.server4Config(config)
            if server_instance:
                servers_dict[server_name] = server_instance

    return servers_dict

rdf_dataset

Created on 2025-05-30

@author: wf

RdfDataset dataclass

Configuration for an RDF dataset to be downloaded.

Source code in omnigraph/rdf_dataset.py 
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
@dataclass
class RdfDataset:
    """
    Configuration for an RDF dataset to be downloaded.
    """

    name: str  # Human-readable dataset name
    base_url: Optional[str] = None  # Base URL e.g. for tryit
    endpoint_url: Optional[str] = None  # SPARQL endpoint URL
    description: Optional[str] = None  # Optional dataset description
    database: Optional[str] = "jena"  # the database type of the endpoint
    expected_solutions: Optional[int] = None  # Expected number of solutions
    select_pattern: str = "?s ?p ?o"  # Basic Graph Pattern for queries
    construct_template: Optional[str] = field(default="?s ?p ?o")
    params: Optional[Dict[str, str]] = None  # npq parameters for {{ name }} templates (issue #36)
    prefix_sets: Optional[list] = field(default_factory=list)
    active: Optional[bool] = False
    rdf_file: Optional[str] = None  # Path to local RDF file (alternative to endpoint_url)
    # fields to be configured by post_init
    id: Optional[str] = field(default=None)
    count_query: Optional[Query] = field(default=None)
    select_query: Optional[Query] = field(default=None)
    sparql: Optional[SPARQL] = field(default=None)

    def apply_params(self, params_dict: Optional[Dict[str, str]] = None) -> None:
        """
        Apply npq parameters to select_pattern and construct_template.

        {{ name }} templates are replaced with the values of the merged
        params: the dataset's own params overridden by the given ones.
        Substitution always starts from the raw templates, so CLI overrides
        can be re-applied after YAML defaults. The blanket Params audit is
        off - it rejects quoted VALUES lists (see issue #36); values stem
        from local configuration and CLI.

        Args:
            params_dict: additional parameter values overriding self.params
        """
        # preserve the raw templates on first call
        if not hasattr(self, "raw_templates"):
            self.raw_templates = {
                "select_pattern": self.select_pattern,
                "construct_template": self.construct_template,
            }
        merged = dict(self.params) if self.params else {}
        if params_dict:
            merged.update(params_dict)
        if merged:
            for attr, raw_template in self.raw_templates.items():
                if raw_template:
                    params = Params(raw_template, with_audit=False)
                    if params.has_params:
                        params.set(merged)
                        setattr(self, attr, params.apply_parameters())

    def build_queries(self) -> None:
        """
        (Re)build count_query and select_query from the current select_pattern.
        Only initializes SPARQL-related fields if endpoint_url is provided.
        """
        if self.endpoint_url:
            self.count_query = Query(
                name=f"{self.name}_count",
                query=f"SELECT (COUNT(*) AS ?count) WHERE {{ {self.select_pattern} }}",
                endpoint=self.endpoint_url,
                description=f"Count query for {self.name}",
            )
            self.select_query = Query(
                name=f"{self.name}_select",
                query=f"SELECT * WHERE {{ {self.select_pattern} }}",
                endpoint=self.endpoint_url,
                description=f"Select query for {self.name}",
            )
            self.sparql = SPARQL(self.endpoint_url)

    def __post_init__(self):
        """
        Apply npq params and generate the queries from select_pattern.
        """
        self.apply_params()
        self.build_queries()

    @property
    def full_name(self):
        ds_id = self.id or "?"
        full_name = f"{ds_id}→{self.name}({self.description})"
        return full_name

    def get_solution_count(self) -> int:
        """
        Get the number of solutions/results from the SPARQL endpoint.

        Returns:
            Number of solutions available from the count query
        """
        count = self.sparql.getValue(self.count_query.query, "count")
        # issue #37: getValue returns the raw literal which may be a str
        count = int(count)
        return count

    def getTryItUrl(self, database: str = "blazegraph") -> str:
        """
        return the "try it!" url for the given database

        Args:
            database(str): the database to be used

        Returns:
            str: the "try it!" url for the given query
        """
        tryit_url = self.select_query.getTryItUrl(self.base_url, database)
        return tryit_url

    def get_construct_query(self, offset: int, limit: int) -> str:
        """
        Generate CONSTRUCT query with offset and limit.

        Args:
            offset: Query offset
            limit: Query limit

        Returns:
            SPARQL CONSTRUCT query string
        """
        query = f"""
        CONSTRUCT {{ {self.construct_template} }}
        WHERE     {{ {self.select_pattern} }}
        OFFSET {offset}
        LIMIT {limit}
        """
        return query

__post_init__()

Apply npq params and generate the queries from select_pattern.

Source code in omnigraph/rdf_dataset.py 
91
92
93
94
95
96
def __post_init__(self):
    """
    Apply npq params and generate the queries from select_pattern.
    """
    self.apply_params()
    self.build_queries()

apply_params(params_dict=None)

Apply npq parameters to select_pattern and construct_template.

{{ name }} templates are replaced with the values of the merged params: the dataset's own params overridden by the given ones. Substitution always starts from the raw templates, so CLI overrides can be re-applied after YAML defaults. The blanket Params audit is off - it rejects quoted VALUES lists (see issue #36); values stem from local configuration and CLI.

Parameters:

Name Type Description Default
params_dict Optional[Dict[str, str]]

additional parameter values overriding self.params

 None
Source code in omnigraph/rdf_dataset.py 
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
def apply_params(self, params_dict: Optional[Dict[str, str]] = None) -> None:
    """
    Apply npq parameters to select_pattern and construct_template.

    {{ name }} templates are replaced with the values of the merged
    params: the dataset's own params overridden by the given ones.
    Substitution always starts from the raw templates, so CLI overrides
    can be re-applied after YAML defaults. The blanket Params audit is
    off - it rejects quoted VALUES lists (see issue #36); values stem
    from local configuration and CLI.

    Args:
        params_dict: additional parameter values overriding self.params
    """
    # preserve the raw templates on first call
    if not hasattr(self, "raw_templates"):
        self.raw_templates = {
            "select_pattern": self.select_pattern,
            "construct_template": self.construct_template,
        }
    merged = dict(self.params) if self.params else {}
    if params_dict:
        merged.update(params_dict)
    if merged:
        for attr, raw_template in self.raw_templates.items():
            if raw_template:
                params = Params(raw_template, with_audit=False)
                if params.has_params:
                    params.set(merged)
                    setattr(self, attr, params.apply_parameters())

build_queries()

(Re)build count_query and select_query from the current select_pattern. Only initializes SPARQL-related fields if endpoint_url is provided.

Source code in omnigraph/rdf_dataset.py 
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
def build_queries(self) -> None:
    """
    (Re)build count_query and select_query from the current select_pattern.
    Only initializes SPARQL-related fields if endpoint_url is provided.
    """
    if self.endpoint_url:
        self.count_query = Query(
            name=f"{self.name}_count",
            query=f"SELECT (COUNT(*) AS ?count) WHERE {{ {self.select_pattern} }}",
            endpoint=self.endpoint_url,
            description=f"Count query for {self.name}",
        )
        self.select_query = Query(
            name=f"{self.name}_select",
            query=f"SELECT * WHERE {{ {self.select_pattern} }}",
            endpoint=self.endpoint_url,
            description=f"Select query for {self.name}",
        )
        self.sparql = SPARQL(self.endpoint_url)

getTryItUrl(database='blazegraph')

return the "try it!" url for the given database

Parameters:

Name Type Description Default
database(str)

the database to be used

 required

Returns:

Name Type Description
str str

the "try it!" url for the given query
Source code in omnigraph/rdf_dataset.py 
116
117
118
119
120
121
122
123
124
125
126
127
def getTryItUrl(self, database: str = "blazegraph") -> str:
    """
    return the "try it!" url for the given database

    Args:
        database(str): the database to be used

    Returns:
        str: the "try it!" url for the given query
    """
    tryit_url = self.select_query.getTryItUrl(self.base_url, database)
    return tryit_url

get_construct_query(offset, limit)

Generate CONSTRUCT query with offset and limit.

Parameters:

Name Type Description Default
offset int

Query offset

 required
limit int

Query limit

 required

Returns:

Type Description
str

SPARQL CONSTRUCT query string
Source code in omnigraph/rdf_dataset.py 
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
def get_construct_query(self, offset: int, limit: int) -> str:
    """
    Generate CONSTRUCT query with offset and limit.

    Args:
        offset: Query offset
        limit: Query limit

    Returns:
        SPARQL CONSTRUCT query string
    """
    query = f"""
    CONSTRUCT {{ {self.construct_template} }}
    WHERE     {{ {self.select_pattern} }}
    OFFSET {offset}
    LIMIT {limit}
    """
    return query

get_solution_count()

Get the number of solutions/results from the SPARQL endpoint.

Returns:

Type Description
int

Number of solutions available from the count query
Source code in omnigraph/rdf_dataset.py 
104
105
106
107
108
109
110
111
112
113
114
def get_solution_count(self) -> int:
    """
    Get the number of solutions/results from the SPARQL endpoint.

    Returns:
        Number of solutions available from the count query
    """
    count = self.sparql.getValue(self.count_query.query, "count")
    # issue #37: getValue returns the raw literal which may be a str
    count = int(count)
    return count

RdfDatasets

Collection of server configurations loaded from YAML.

Source code in omnigraph/rdf_dataset.py 
149
150
151
152
153
154
155
156
157
158
159
160
161
@lod_storable
class RdfDatasets:
    """Collection of server configurations loaded from YAML."""

    datasets: Dict[str, RdfDataset] = field(default_factory=dict)

    @classmethod
    def ofYaml(cls, yaml_path: str) -> "RdfDatasets":
        """Load server configurations from YAML file."""
        datasets = cls.load_from_yaml_file(yaml_path)
        for ds_id, dataset in datasets.datasets.items():
            dataset.id = ds_id
        return datasets

ofYaml(yaml_path) classmethod

Load server configurations from YAML file.

Source code in omnigraph/rdf_dataset.py 
155
156
157
158
159
160
161
@classmethod
def ofYaml(cls, yaml_path: str) -> "RdfDatasets":
    """Load server configurations from YAML file."""
    datasets = cls.load_from_yaml_file(yaml_path)
    for ds_id, dataset in datasets.datasets.items():
        dataset.id = ds_id
    return datasets

rdfdump

Created on 2025-05-26

@author: wf

Download RDF dump via paginated CONSTRUCT queries.

RdfDumpDownloader

Downloads an RDF dump from a SPARQL endpoint via paginated CONSTRUCT queries.

Source code in omnigraph/rdfdump.py 
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
class RdfDumpDownloader:
    """
    Downloads an RDF dump from a SPARQL endpoint via
    paginated CONSTRUCT queries.
    """

    def __init__(self, dataset: RdfDataset, output_path: str, args: Optional[Namespace] = None):
        """
        Initialize the RDF dump downloader.

        Args:
            dataset: RdfDataset configuration
            output_path: the directory for the dump file
            args: parsed CLI arguments (optional)
        """
        self.args = args
        self.rdf_format = RdfFormat.by_label(args.rdf_format)
        self.dataset = dataset
        self.endpoint_url = dataset.endpoint_url
        self.sparql = SPARQL(self.endpoint_url)
        self.output_path = output_path
        self.limit = args.limit if args else 10000
        self.max_count = args.max_count if args and args.max_count is not None else dataset.expected_solutions or 200000
        self.show_progress = not args.no_progress if args else True
        self.force = args.force if args else False
        self.debug = args.debug if args else False

    def fetch_chunk(self, offset: int, rdf_format: str = "turtle") -> str:
        """
        Fetch a chunk of RDF data in the given format using direct HTTP POST.

        Args:
            offset: Query offset
            rdf_format: RDF format label

        Returns:
            RDF content as string
        """
        query = self.dataset.get_construct_query(offset, self.limit)
        if self.debug:
            print(query)
        content = self.sparql.post_query_direct(query=query, rdf_format=rdf_format)
        # Better debugging
        if self.debug:
            print(f"Chunk {offset}: content length = {len(content) if content else 0}")
        return content

    def download(self) -> int:
        """
        Download the RDF dump in chunks.

        Returns:
            Number of chunks downloaded
        """
        # make sure the output_path is created
        output_dir = Path(self.output_path)
        output_dir.mkdir(parents=True, exist_ok=True)

        # Get actual count from dataset
        actual_count = self.dataset.get_solution_count()
        total_chunks = (actual_count + self.limit - 1) // self.limit  # Round up
        chunk_count = 0

        iterator = range(total_chunks)
        if self.show_progress:
            iterator = tqdm(iterator, desc=f"Downloading RDF dump ({actual_count} results)")

        for chunk_idx in iterator:
            filename = output_dir / f"dump_{chunk_idx:06d}{self.rdf_format.extension}"
            if filename.exists() and not self.force:
                if self.show_progress:
                    iterator.set_description(f"Skipping existing file: {filename}")
                continue

            offset = chunk_idx * self.limit
            try:
                content = self.fetch_chunk(offset=offset, rdf_format=self.rdf_format.label)
            except Exception as e:
                print(f"Error at offset {offset}: {e}")
                break

            if not content or content.strip() == "":
                break

            with open(filename, "w", encoding="utf-8") as f:
                f.write(content)

            chunk_count += 1
            time.sleep(0.5)

        return chunk_count

__init__(dataset, output_path, args=None)

Initialize the RDF dump downloader.

Parameters:

Name Type Description Default
dataset RdfDataset

RdfDataset configuration

 required
output_path str

the directory for the dump file

 required
args Optional[Namespace]

parsed CLI arguments (optional)

 None
Source code in omnigraph/rdfdump.py 
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
def __init__(self, dataset: RdfDataset, output_path: str, args: Optional[Namespace] = None):
    """
    Initialize the RDF dump downloader.

    Args:
        dataset: RdfDataset configuration
        output_path: the directory for the dump file
        args: parsed CLI arguments (optional)
    """
    self.args = args
    self.rdf_format = RdfFormat.by_label(args.rdf_format)
    self.dataset = dataset
    self.endpoint_url = dataset.endpoint_url
    self.sparql = SPARQL(self.endpoint_url)
    self.output_path = output_path
    self.limit = args.limit if args else 10000
    self.max_count = args.max_count if args and args.max_count is not None else dataset.expected_solutions or 200000
    self.show_progress = not args.no_progress if args else True
    self.force = args.force if args else False
    self.debug = args.debug if args else False

download()

Download the RDF dump in chunks.

Returns:

Type Description
int

Number of chunks downloaded
Source code in omnigraph/rdfdump.py 
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
def download(self) -> int:
    """
    Download the RDF dump in chunks.

    Returns:
        Number of chunks downloaded
    """
    # make sure the output_path is created
    output_dir = Path(self.output_path)
    output_dir.mkdir(parents=True, exist_ok=True)

    # Get actual count from dataset
    actual_count = self.dataset.get_solution_count()
    total_chunks = (actual_count + self.limit - 1) // self.limit  # Round up
    chunk_count = 0

    iterator = range(total_chunks)
    if self.show_progress:
        iterator = tqdm(iterator, desc=f"Downloading RDF dump ({actual_count} results)")

    for chunk_idx in iterator:
        filename = output_dir / f"dump_{chunk_idx:06d}{self.rdf_format.extension}"
        if filename.exists() and not self.force:
            if self.show_progress:
                iterator.set_description(f"Skipping existing file: {filename}")
            continue

        offset = chunk_idx * self.limit
        try:
            content = self.fetch_chunk(offset=offset, rdf_format=self.rdf_format.label)
        except Exception as e:
            print(f"Error at offset {offset}: {e}")
            break

        if not content or content.strip() == "":
            break

        with open(filename, "w", encoding="utf-8") as f:
            f.write(content)

        chunk_count += 1
        time.sleep(0.5)

    return chunk_count

fetch_chunk(offset, rdf_format='turtle')

Fetch a chunk of RDF data in the given format using direct HTTP POST.

Parameters:

Name Type Description Default
offset int

Query offset

 required
rdf_format str

RDF format label

 'turtle'

Returns:

Type Description
str

RDF content as string
Source code in omnigraph/rdfdump.py 
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
def fetch_chunk(self, offset: int, rdf_format: str = "turtle") -> str:
    """
    Fetch a chunk of RDF data in the given format using direct HTTP POST.

    Args:
        offset: Query offset
        rdf_format: RDF format label

    Returns:
        RDF content as string
    """
    query = self.dataset.get_construct_query(offset, self.limit)
    if self.debug:
        print(query)
    content = self.sparql.post_query_direct(query=query, rdf_format=rdf_format)
    # Better debugging
    if self.debug:
        print(f"Chunk {offset}: content length = {len(content) if content else 0}")
    return content

rdfdump_cmd

Created on 2025-05-30

@author: wf

Command line interface for RDF dump downloading.

RdfDumpCmd

Bases: BaseCmd

Command line interface for RDF dump downloading.

Source code in omnigraph/rdfdump_cmd.py 
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
class RdfDumpCmd(BaseCmd):
    """
    Command line interface for RDF dump downloading.
    """

    def __init__(self):
        """
        Initialize command line interface.
        """
        super().__init__(description="Download RDF dump from SPARQL endpoint via paginated CONSTRUCT queries")

    def get_arg_parser(self, description: str, version_msg: str) -> ArgumentParser:
        """
        Extend base parser with RDF-specific arguments.

        Args:
            description: CLI description string
            version_msg: version display string

        Returns:
            ArgumentParser: extended argument parser
        """
        parser = super().get_arg_parser(description, version_msg)
        parser.add_argument(
            "--limit",
            type=int,
            default=10000,
            help="Number of triples per request [default: %(default)s]",
        )
        parser.add_argument("-l", "--list", action="store_true", help="List available datasets [default: %(default)s]")
        parser.add_argument(
            "--count", action="store_true", help="List available datasets with triple counts[default: %(default)s]"
        )
        parser.add_argument("--dump", action="store_true", help="perform the dump [default: %(default)s]")
        parser.add_argument(
            "-4o",
            "--for-omnigraph",
            action="store_true",
            help="store dump at default omnigraph location [default: %(default)s]",
        )
        parser.add_argument(
            "--max-count",
            type=int,
            default=None,
            help="Maximum number of solutions/triples to download (uses dataset expected_solutions if not specified)",
        )
        parser.add_argument(
            "--param",
            action=StoreDictKeyPair,
            dest="params",
            metavar="KEY=VALUE,KEY=VALUE...",
            help="npq parameter values overriding the dataset's params (issue #36)",
        )
        parser.add_argument("--no-progress", action="store_true", help="Disable progress bar")
        parser.add_argument("--output-path", default=".", help="Path for dump files")
        parser.add_argument("--tryit", action="store_true", help="open the try it! URL [default: %(default)s]")

        return parser

    def download_dataset(self, dataset_name: str, dataset: RdfDataset, output_path: str):
        """
        Download the specified dataset to a subdirectory.

        Args:
            dataset_name: name of dataset
            dataset: RDF dataset definition
            output_path: base output directory
        """
        dataset_dir = os.path.join(output_path, dataset_name)
        os.makedirs(dataset_dir, exist_ok=True)
        if not self.quiet:
            print(
                f"Starting download for dataset: {dataset_name} to {dataset_dir} in {self.rdf_format.label} format ..."
            )

        downloader = RdfDumpDownloader(dataset=dataset, output_path=dataset_dir, args=self.args)

        chunk_count = downloader.download()
        print(f"Dataset {dataset_name}: Downloaded {chunk_count} {self.rdf_format.extension} files.")

    def handle_args(self, args: Namespace):
        """
        Handle parsed CLI arguments.

        Args:
            args: parsed namespace
        """
        super().handle_args(args)
        datasets = self.datasets
        if getattr(args, "params", None):
            # apply CLI npq parameter overrides and rebuild the queries
            for dataset in datasets.values():
                dataset.apply_params(args.params)
                dataset.build_queries()
        if self.args.about:
            self.about()

        if self.args.list:
            print("Available datasets:")
            for dataset in self.all_datasets.datasets.values():
                print(f"  {dataset.full_name}")
            return

        if self.args.count:
            print("Triple count for available datasets:")
            for dataset in datasets.values():
                tryit_url = dataset.getTryItUrl(dataset.database)
                print(f"  {dataset.full_name}")
                if self.args.tryit:
                    webbrowser.open(tryit_url)
                count = dataset.sparql.getValue(dataset.count_query.query, "count")
                print(f"  {count} triples")

        output_path = self.args.output_path
        if self.args.for_omnigraph:
            output_path = self.ogp.dumps_dir

        if self.args.dump:
            for dataset_name, dataset in datasets.items():
                self.download_dataset(dataset_name, dataset, output_path)

__init__()

Initialize command line interface.

Source code in omnigraph/rdfdump_cmd.py 
25
26
27
28
29
def __init__(self):
    """
    Initialize command line interface.
    """
    super().__init__(description="Download RDF dump from SPARQL endpoint via paginated CONSTRUCT queries")

download_dataset(dataset_name, dataset, output_path)

Download the specified dataset to a subdirectory.

Parameters:

Name Type Description Default
dataset_name str

name of dataset

 required
dataset RdfDataset

RDF dataset definition

 required
output_path str

base output directory

 required
Source code in omnigraph/rdfdump_cmd.py 
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
def download_dataset(self, dataset_name: str, dataset: RdfDataset, output_path: str):
    """
    Download the specified dataset to a subdirectory.

    Args:
        dataset_name: name of dataset
        dataset: RDF dataset definition
        output_path: base output directory
    """
    dataset_dir = os.path.join(output_path, dataset_name)
    os.makedirs(dataset_dir, exist_ok=True)
    if not self.quiet:
        print(
            f"Starting download for dataset: {dataset_name} to {dataset_dir} in {self.rdf_format.label} format ..."
        )

    downloader = RdfDumpDownloader(dataset=dataset, output_path=dataset_dir, args=self.args)

    chunk_count = downloader.download()
    print(f"Dataset {dataset_name}: Downloaded {chunk_count} {self.rdf_format.extension} files.")

get_arg_parser(description, version_msg)

Extend base parser with RDF-specific arguments.

Parameters:

Name Type Description Default
description str

CLI description string

 required
version_msg str

version display string

 required

Returns:

Name Type Description
ArgumentParser ArgumentParser

extended argument parser
Source code in omnigraph/rdfdump_cmd.py 
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
def get_arg_parser(self, description: str, version_msg: str) -> ArgumentParser:
    """
    Extend base parser with RDF-specific arguments.

    Args:
        description: CLI description string
        version_msg: version display string

    Returns:
        ArgumentParser: extended argument parser
    """
    parser = super().get_arg_parser(description, version_msg)
    parser.add_argument(
        "--limit",
        type=int,
        default=10000,
        help="Number of triples per request [default: %(default)s]",
    )
    parser.add_argument("-l", "--list", action="store_true", help="List available datasets [default: %(default)s]")
    parser.add_argument(
        "--count", action="store_true", help="List available datasets with triple counts[default: %(default)s]"
    )
    parser.add_argument("--dump", action="store_true", help="perform the dump [default: %(default)s]")
    parser.add_argument(
        "-4o",
        "--for-omnigraph",
        action="store_true",
        help="store dump at default omnigraph location [default: %(default)s]",
    )
    parser.add_argument(
        "--max-count",
        type=int,
        default=None,
        help="Maximum number of solutions/triples to download (uses dataset expected_solutions if not specified)",
    )
    parser.add_argument(
        "--param",
        action=StoreDictKeyPair,
        dest="params",
        metavar="KEY=VALUE,KEY=VALUE...",
        help="npq parameter values overriding the dataset's params (issue #36)",
    )
    parser.add_argument("--no-progress", action="store_true", help="Disable progress bar")
    parser.add_argument("--output-path", default=".", help="Path for dump files")
    parser.add_argument("--tryit", action="store_true", help="open the try it! URL [default: %(default)s]")

    return parser

handle_args(args)

Handle parsed CLI arguments.

Parameters:

Name Type Description Default
args Namespace

parsed namespace

 required
Source code in omnigraph/rdfdump_cmd.py 
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
def handle_args(self, args: Namespace):
    """
    Handle parsed CLI arguments.

    Args:
        args: parsed namespace
    """
    super().handle_args(args)
    datasets = self.datasets
    if getattr(args, "params", None):
        # apply CLI npq parameter overrides and rebuild the queries
        for dataset in datasets.values():
            dataset.apply_params(args.params)
            dataset.build_queries()
    if self.args.about:
        self.about()

    if self.args.list:
        print("Available datasets:")
        for dataset in self.all_datasets.datasets.values():
            print(f"  {dataset.full_name}")
        return

    if self.args.count:
        print("Triple count for available datasets:")
        for dataset in datasets.values():
            tryit_url = dataset.getTryItUrl(dataset.database)
            print(f"  {dataset.full_name}")
            if self.args.tryit:
                webbrowser.open(tryit_url)
            count = dataset.sparql.getValue(dataset.count_query.query, "count")
            print(f"  {count} triples")

    output_path = self.args.output_path
    if self.args.for_omnigraph:
        output_path = self.ogp.dumps_dir

    if self.args.dump:
        for dataset_name, dataset in datasets.items():
            self.download_dataset(dataset_name, dataset, output_path)

server_config

Created on 2025-05-28

@author: wf

ServerCmd dataclass

Command wrapper for server operations.

Source code in omnigraph/server_config.py 
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
@dataclass
class ServerCmd:
    """
    Command wrapper for server operations.
    """

    def __init__(self, title: str, func: Callable):
        """
        Initialize server command.

        Args:
            title: Description of the command
            func: Function to execute
        """
        self.title = title
        self.func = func

    def run(self, verbose: bool = True) -> any:
        """
        Execute the server command.

        Args:
            verbose: Whether to print result

        Returns:
            Result from function execution
        """
        if verbose:
            print(f"{self.title} ...")
        result = self.func()
        if verbose:
            print(f"{self.title}: {result}")
        return result

__init__(title, func)

Initialize server command.

Parameters:

Name Type Description Default
title str

Description of the command

 required
func Callable

Function to execute

 required
Source code in omnigraph/server_config.py 
343
344
345
346
347
348
349
350
351
352
def __init__(self, title: str, func: Callable):
    """
    Initialize server command.

    Args:
        title: Description of the command
        func: Function to execute
    """
    self.title = title
    self.func = func

run(verbose=True)

Execute the server command.

Parameters:

Name Type Description Default
verbose bool

Whether to print result

 True

Returns:

Type Description
any

Result from function execution
Source code in omnigraph/server_config.py 
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
def run(self, verbose: bool = True) -> any:
    """
    Execute the server command.

    Args:
        verbose: Whether to print result

    Returns:
        Result from function execution
    """
    if verbose:
        print(f"{self.title} ...")
    result = self.func()
    if verbose:
        print(f"{self.title}: {result}")
    return result

ServerConfig dataclass

a server configuration for a Knowledge Graph endpoint potentially provided by a docker container and often implemented as a SPARQL endpoint

Source code in omnigraph/server_config.py 
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
@dataclass
class ServerConfig:
    """
    a server configuration for a Knowledge Graph endpoint
    potentially provided by a docker container and often
    implemented as a SPARQL endpoint
    """

    server: str
    name: str
    wikidata_id: str
    container_name: str
    port: int
    test_port: int
    image: str
    free_image: Optional[str] = None  # Fallback image if license is missing
    license_env_var: Optional[str] = None  # Name of EnvVar holding the key
    active: bool = True
    has_license: bool = False
    protocol: str = "http"
    host: str = "localhost"
    docker_bind: str = "127.0.0.1"  # Docker port bind address (default: localhost-only for security)
    rdf_format: str = "turtle"
    auth_user: Optional[str] = None
    auth_password: Optional[str] = None
    dataset: Optional[str] = None
    prefix_sets: Optional[list] = field(default_factory=lambda: ["rdf"])
    timeout: int = 30
    ready_timeout: int = 20
    proxy_timeout: int = 5400  # e.g. apache server
    upload_timeout: int = 300
    unforced_clear_limit = 100000  # maximumn number of triples that can be cleared without force option
    # fields to be configured by post_init
    base_url: Optional[str] = field(default=None)
    status_url: Optional[str] = field(default=None)
    web_url: Optional[str] = field(default=None)
    sparql_url: Optional[str] = field(default=None)
    upload_url: Optional[str] = field(default=None)
    base_data_dir: Optional[str] = field(default=None)  # base data directory available as bind mount
    data_dir: Optional[str] = field(default=None)  # default data directory
    dumps_dir: Optional[str] = field(default=None)
    needed_software: Optional[SoftwareList] = field(default=None)

    def __post_init__(self):
        if self.base_url is None:
            self.base_url = f"{self.protocol}://{self.host}:{self.port}"
        # Check if we have a license available
        self.has_license = self.license_env_var and os.environ.get(self.license_env_var)

    @property
    def support_status(self) -> SupportStatus:
        """
        Determine support status based on configuration and environment.

        Returns:
            SupportStatus enum indicating if/how server can run
        """
        status = SupportStatus.SUPPORTED

        # Check if manually disabled
        if not self.active:
            status = SupportStatus.MANUAL_DISABLED
        elif self.license_env_var:
            # Check license requirements
            if self.has_license:
                status = SupportStatus.SUPPORTED
            elif self.free_image:
                status = SupportStatus.LIMITED
            else:
                status = SupportStatus.MISSING_LICENSE

        return status

    @property
    def effective_image(self) -> str:
        # Default to the main image
        target_image = self.image

        # Downgrade to free image ONLY if we have no license AND a free alternative exists
        if self.free_image and not self.has_license:
            target_image = self.free_image
            if target_image.startswith("https://github.com/"):
                target_image = target_image.split("/")[-1].replace(".git", "") + ":local"
        return target_image

    @property
    def docker_user_flag(self) -> str:
        try:
            uid = os.getuid()
            gid = os.getgid()
            user_flag = f"-u {uid}:{gid}"
        except AttributeError:
            # e.g. on Windows
            user_flag = ""
        return user_flag

    def generator_header(self, version=None) -> str:
        """
        generate a standard header with timestamp and optional version information

        Args:
            version: optional version info, defaults to Version.version

        Returns:
            str: a header string suitable for generated files
        """
        iso_timestamp = datetime.now().isoformat()
        version_info = ""
        if version is None:
            version = Version
        if version:
            version_info = f"""{version.name} Version {version.version} of {version.updated} ({version.description})"""

        header = f"""# Generated by omnigraph at {iso_timestamp}
# {version_info}"""
        return header

    def to_apache_config(self, domain: str, version: None) -> str:
        """
        Generate Apache configuration based for this server.

        Args:
            domain(str): the base domain to use
            version: the omnigraph Version info to use
        Returns:
            str: The Apache configuration as a string.
        """
        server_name = f"{self.name}.{domain}"
        admin_email = f"webmaster@{domain}"
        header = self.generator_header(version)
        header_comment = f"""# Apache Configuration for {server_name}
# {header}
# http Port: {self.port}
# SSL Port: 443
# timeout: {self.proxy_timeout}
"""

        template = """<VirtualHost *:{port}>
    ServerName {server_name}
    ServerAdmin {admin_email}

    {ssl_config_part}
    ErrorLog ${{APACHE_LOG_DIR}}/{short_name}_error{log_suffix}.log
    CustomLog ${{APACHE_LOG_DIR}}/{short_name}{log_suffix}.log combined

    ProxyPreserveHost On
    ProxyTimeout {proxy_timeout}

    ProxyPass / http://localhost:{default_port}/
    ProxyPassReverse / http://localhost:{default_port}/
</VirtualHost>
"""

        # For SSL Configuration
        ssl_config = template.format(
            port=443,
            server_name=server_name,
            admin_email=admin_email,
            short_name=self.name,
            log_suffix="_ssl",
            default_port=self.port,
            proxy_timeout=self.proxy_timeout,
            ssl_config_part="Include ssl.conf",
        )

        # For Non-SSL Configuration
        http_config = template.format(
            port=80,
            server_name=server_name,
            admin_email=admin_email,
            short_name=self.name,
            log_suffix="",
            default_port=self.port,
            proxy_timeout=self.proxy_timeout,
            ssl_config_part="",
        )

        apache_config = header_comment + ssl_config + http_config
        return apache_config

support_status property

Determine support status based on configuration and environment.

Returns:

Type Description
SupportStatus

SupportStatus enum indicating if/how server can run

generator_header(version=None)

generate a standard header with timestamp and optional version information

Parameters:

Name Type Description Default
version

optional version info, defaults to Version.version

 None

Returns:

Name Type Description
str str

a header string suitable for generated files
Source code in omnigraph/server_config.py 
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
    def generator_header(self, version=None) -> str:
        """
        generate a standard header with timestamp and optional version information

        Args:
            version: optional version info, defaults to Version.version

        Returns:
            str: a header string suitable for generated files
        """
        iso_timestamp = datetime.now().isoformat()
        version_info = ""
        if version is None:
            version = Version
        if version:
            version_info = f"""{version.name} Version {version.version} of {version.updated} ({version.description})"""

        header = f"""# Generated by omnigraph at {iso_timestamp}
# {version_info}"""
        return header

to_apache_config(domain, version)

Generate Apache configuration based for this server.

Parameters:

Name Type Description Default
domain(str)

the base domain to use

 required
version None

the omnigraph Version info to use

 required

Returns: str: The Apache configuration as a string.

Source code in omnigraph/server_config.py 
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
    def to_apache_config(self, domain: str, version: None) -> str:
        """
        Generate Apache configuration based for this server.

        Args:
            domain(str): the base domain to use
            version: the omnigraph Version info to use
        Returns:
            str: The Apache configuration as a string.
        """
        server_name = f"{self.name}.{domain}"
        admin_email = f"webmaster@{domain}"
        header = self.generator_header(version)
        header_comment = f"""# Apache Configuration for {server_name}
# {header}
# http Port: {self.port}
# SSL Port: 443
# timeout: {self.proxy_timeout}
"""

        template = """<VirtualHost *:{port}>
    ServerName {server_name}
    ServerAdmin {admin_email}

    {ssl_config_part}
    ErrorLog ${{APACHE_LOG_DIR}}/{short_name}_error{log_suffix}.log
    CustomLog ${{APACHE_LOG_DIR}}/{short_name}{log_suffix}.log combined

    ProxyPreserveHost On
    ProxyTimeout {proxy_timeout}

    ProxyPass / http://localhost:{default_port}/
    ProxyPassReverse / http://localhost:{default_port}/
</VirtualHost>
"""

        # For SSL Configuration
        ssl_config = template.format(
            port=443,
            server_name=server_name,
            admin_email=admin_email,
            short_name=self.name,
            log_suffix="_ssl",
            default_port=self.port,
            proxy_timeout=self.proxy_timeout,
            ssl_config_part="Include ssl.conf",
        )

        # For Non-SSL Configuration
        http_config = template.format(
            port=80,
            server_name=server_name,
            admin_email=admin_email,
            short_name=self.name,
            log_suffix="",
            default_port=self.port,
            proxy_timeout=self.proxy_timeout,
            ssl_config_part="",
        )

        apache_config = header_comment + ssl_config + http_config
        return apache_config

ServerConfigs

Collection of server configurations loaded from YAML.

Source code in omnigraph/server_config.py 
324
325
326
327
328
329
330
331
332
333
334
@lod_storable
class ServerConfigs:
    """Collection of server configurations loaded from YAML."""

    servers: Dict[str, ServerConfig] = field(default_factory=dict)

    @classmethod
    def ofYaml(cls, yaml_path: str) -> "ServerConfigs":
        """Load server configurations from YAML file."""
        server_configs = cls.load_from_yaml_file(yaml_path)
        return server_configs

ofYaml(yaml_path) classmethod

Load server configurations from YAML file.

Source code in omnigraph/server_config.py 
330
331
332
333
334
@classmethod
def ofYaml(cls, yaml_path: str) -> "ServerConfigs":
    """Load server configurations from YAML file."""
    server_configs = cls.load_from_yaml_file(yaml_path)
    return server_configs

ServerEnv

Server environment configuration.

Source code in omnigraph/server_config.py 
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
class ServerEnv:
    """
    Server environment configuration.
    """

    def __init__(
        self, log: Log = None, shell: Shell = None, force: bool = False, debug: bool = False, verbose: bool = False
    ):
        """
        Initialize server environment.

        Args:
            log: Log instance for logging
            shell: Shell instance for command execution
            force: if True enable actions that are otherwise protected e.g. deletion of data
            debug: Enable debug mode
            verbose: Enable verbose output
        """
        if log is None:
            log = Log()
            log.do_print = debug and verbose
        self.log = log
        if shell is None:
            shell = Shell()
        self.shell = shell
        self.force = force
        self.debug = debug
        self.verbose = verbose

__init__(log=None, shell=None, force=False, debug=False, verbose=False)

Initialize server environment.

Parameters:

Name Type Description Default
log Log

Log instance for logging

 None
shell Shell

Shell instance for command execution

 None
force bool

if True enable actions that are otherwise protected e.g. deletion of data

 False
debug bool

Enable debug mode

 False
verbose bool

Enable verbose output

 False
Source code in omnigraph/server_config.py 
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
def __init__(
    self, log: Log = None, shell: Shell = None, force: bool = False, debug: bool = False, verbose: bool = False
):
    """
    Initialize server environment.

    Args:
        log: Log instance for logging
        shell: Shell instance for command execution
        force: if True enable actions that are otherwise protected e.g. deletion of data
        debug: Enable debug mode
        verbose: Enable verbose output
    """
    if log is None:
        log = Log()
        log.do_print = debug and verbose
    self.log = log
    if shell is None:
        shell = Shell()
    self.shell = shell
    self.force = force
    self.debug = debug
    self.verbose = verbose

ServerLifecycleState

Bases: Enum

a state in the servers lifecycle

Source code in omnigraph/server_config.py 
61
62
63
64
65
66
67
68
69
70
71
class ServerLifecycleState(Enum):
    """
    a state in the servers lifecycle
    """

    READY = "ready ✅"
    UP = "up 🟢"
    ERROR = "error ❌"
    UNKNOWN = "unknown ❓"
    STARTING = "starting 🔄"
    STOPPED = "stopped ⏹️"

ServerStatus dataclass

Server status

Source code in omnigraph/server_config.py 
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
@dataclass
class ServerStatus:
    """
    Server status
    """

    at: ServerLifecycleState
    running: bool = False
    exists: bool = False
    error: Optional[Exception] = None
    http_status_code: Optional[int] = None
    docker_status: Optional[str] = None
    docker_exit_code: Optional[int] = None
    # fields to be initialized by post_init
    logs: str = field(default=None)
    triple_count: int = field(default=None)
    timestamp: datetime = field(default=None)
    status_dict: Dict[str, str] = field(default_factory=dict)

    def __post_init__(self):
        self.timestamp = datetime.now()

    def get_summary(self, debug: bool) -> str:
        """
        get a summary of the Server Status
        """
        summary = f"@ {self.timestamp.strftime('%H:%M:%S')}"
        if self.http_status_code:
            summary += f" (HTTP {self.http_status_code})"
        if self.triple_count:
            summary += f"{self.triple_count} triples"
        if self.error:
            debug_msg = f" - {type(self.error).__name__}"
            if debug:
                debug_msg = "".join(traceback.format_exception(type(self.error), self.error, self.error.__traceback__))
            summary += debug_msg
        return summary

get_summary(debug)

get a summary of the Server Status

Source code in omnigraph/server_config.py 
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
def get_summary(self, debug: bool) -> str:
    """
    get a summary of the Server Status
    """
    summary = f"@ {self.timestamp.strftime('%H:%M:%S')}"
    if self.http_status_code:
        summary += f" (HTTP {self.http_status_code})"
    if self.triple_count:
        summary += f"{self.triple_count} triples"
    if self.error:
        debug_msg = f" - {type(self.error).__name__}"
        if debug:
            debug_msg = "".join(traceback.format_exception(type(self.error), self.error, self.error.__traceback__))
        summary += debug_msg
    return summary

SupportStatus

Bases: Enum

Determines if and how a server can run based on environment/licenses.

Source code in omnigraph/server_config.py 
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
class SupportStatus(Enum):
    """
    Determines if and how a server can run based on environment/licenses.
    """

    SUPPORTED = "supported ✅"
    LIMITED = "limited ⚠️"
    MISSING_LICENSE = "no_license 🛑"
    MISSING_SOFTWARE = "no_software 🛑"
    MANUAL_DISABLED = "disabled ⛔"

    def can_start(self) -> bool:
        """Check if server can be started with this status."""
        return self in (SupportStatus.SUPPORTED, SupportStatus.LIMITED)

    def is_blocking(self) -> bool:
        """Check if this status blocks server operations."""
        return not self.can_start()

    def log_status(self, log: Log, container_name: str, config: "ServerConfig") -> None:
        """Log appropriate message for this status."""
        if self == SupportStatus.MANUAL_DISABLED:
            log.log("🛑", container_name, "Server is manually disabled in configuration")

        elif self == SupportStatus.MISSING_SOFTWARE:
            log.log("🛑", container_name, "Required software missing - cannot start")

        elif self == SupportStatus.MISSING_LICENSE:
            log.log(
                "🛑", container_name, f"License required (set {config.license_env_var}) and no free fallback available"
            )

        elif self == SupportStatus.LIMITED:
            log.log("⚠️", container_name, f"Using free/community image: {config.effective_image}")

        elif self == SupportStatus.SUPPORTED:
            log.log("✅", container_name, "Server fully supported")

can_start()

Check if server can be started with this status.

Source code in omnigraph/server_config.py 
33
34
35
def can_start(self) -> bool:
    """Check if server can be started with this status."""
    return self in (SupportStatus.SUPPORTED, SupportStatus.LIMITED)

is_blocking()

Check if this status blocks server operations.

Source code in omnigraph/server_config.py 
37
38
39
def is_blocking(self) -> bool:
    """Check if this status blocks server operations."""
    return not self.can_start()

log_status(log, container_name, config)

Log appropriate message for this status.

Source code in omnigraph/server_config.py 
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
def log_status(self, log: Log, container_name: str, config: "ServerConfig") -> None:
    """Log appropriate message for this status."""
    if self == SupportStatus.MANUAL_DISABLED:
        log.log("🛑", container_name, "Server is manually disabled in configuration")

    elif self == SupportStatus.MISSING_SOFTWARE:
        log.log("🛑", container_name, "Required software missing - cannot start")

    elif self == SupportStatus.MISSING_LICENSE:
        log.log(
            "🛑", container_name, f"License required (set {config.license_env_var}) and no free fallback available"
        )

    elif self == SupportStatus.LIMITED:
        log.log("⚠️", container_name, f"Using free/community image: {config.effective_image}")

    elif self == SupportStatus.SUPPORTED:
        log.log("✅", container_name, "Server fully supported")

servers

blazegraph

Created on 2025-05-27

@author: wf

Blazegraph

Bases: SparqlServer

Dockerized Blazegraph SPARQL server

Source code in omnigraph/servers/blazegraph.py 
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
class Blazegraph(SparqlServer):
    """
    Dockerized Blazegraph SPARQL server
    """

    def __init__(self, config: ServerConfig, env: ServerEnv):
        """
        Initialize the Blazegraph manager.

        Args:
            config: Server configuration
            env: Server environment (includes log, shell, debug, verbose)
        """
        super().__init__(config=config, env=env)

    def pre_create(self):
        """
        Prepare Blazegraph data directory and RWStore.properties.
        """
        data_dir = Path(self.config.base_data_dir)
        data_dir.mkdir(parents=True, exist_ok=True)
        rwstore_path = data_dir / "RWStore.properties"
        if not rwstore_path.exists():
            header = self.config.generator_header()
            props = f"""{header}

    # Blazegraph journal configuration
    com.bigdata.journal.AbstractJournal.file=/data/blazegraph.jnl

    # Enable text index
    com.bigdata.rdf.store.AbstractTripleStore.textIndex=true

    # No OWL reasoning
    com.bigdata.rdf.store.AbstractTripleStore.axiomsClass=com.bigdata.rdf.axioms.NoAxioms

    # No justification or truth maintenance
    com.bigdata.rdf.sail.truthMaintenance=false
    com.bigdata.rdf.store.AbstractTripleStore.justify=false

    # Default namespace
    com.bigdata.rdf.sail.namespace={self.config.dataset}
    """
            rwstore_path.write_text(props)


    def get_dataloader_xml(self, container_path: str) -> str:
        """
        Build the DataLoader servlet properties XML for the given
        container-side file or directory path.

        Args:
            container_path: path as seen inside the container (mounted /data)

        Returns:
            the properties XML document
        """
        xml = f"""<?xml version="1.0" encoding="UTF-8" standalone="no"?>
<!DOCTYPE properties SYSTEM "http://java.sun.com/dtd/properties.dtd">
<properties>
  <entry key="namespace">{self.config.dataset}</entry>
  <entry key="propertyFile">/RWStore.properties</entry>
  <entry key="fileOrDirs">{container_path}</entry>
  <entry key="-durableQueues">true</entry>
</properties>"""
        return xml

    def upload_dump_files(self, file_pattern: str = None) -> int:
        """
        Bulk-load dump files via Blazegraph's REST DataLoader servlet, which
        reads the files from the server's own filesystem (the mounted /data
        directory) instead of pushing them through HTTP request bodies.

        Args:
            file_pattern: Glob pattern for dump files

        Returns:
            Number of files loaded successfully
        """
        container_name = self.config.container_name
        files = self.get_dump_files(file_pattern)
        loaded_count = 0
        if not files:
            self.log.log("⚠️", container_name, f"No dump files found for pattern: {file_pattern}")
        else:
            # stage the dumps under data_dir so the container sees them at /data/dumps
            stage_dir = Path(self.config.base_data_dir) / "dumps"
            stage_dir.mkdir(parents=True, exist_ok=True)
            for file in files:
                target = stage_dir / file.name
                if not target.exists():
                    shutil.copy2(file, target)
            xml = self.get_dataloader_xml("/data/dumps")
            response = self.make_request(
                "POST",
                self.config.dataloader_url,
                headers={"Content-Type": "application/xml"},
                data=xml,
                timeout=self.config.upload_timeout,
            )
            if response.success:
                loaded_count = len(files)
                self.log.log("✅", container_name, f"DataLoader loaded {loaded_count} file(s)")
            else:
                error_msg = str(response.error) if response.error else f"HTTP {response.response.status_code}"
                self.log.log("❌", container_name, f"DataLoader failed: {error_msg}")
        return loaded_count

    def status(self) -> ServerStatus:
        """
        Get server status information.

        Returns:
            ServerStatus object with status information
        """
        server_status = super().status()
        if server_status.exists and server_status.running:
            response = self.make_request("GET", self.config.status_url)

            if response.success:
                lifecycle = ServerLifecycleState.READY

            if response.response and response.response.text:
                html_content = response.response.text
                # Only parse HTML if it looks like HTML content
                if "<" in html_content and ">" in html_content:
                    name_value_pattern = r'(?:<span id="(?P<name1>[^"]+)">(?P<value1>[^<]+)</span[^>]*>|&#47;(?P<name2>[^=]+)=(?P<value2>[^\s&#]+))'
                    matches = re.finditer(name_value_pattern, html_content, re.DOTALL)

                    for match in matches:
                        for name_group, value_group in {
                            "name1": "value1",
                            "name2": "value2",
                        }.items():
                            name = match.group(name_group)
                            if name:
                                value = match.group(value_group)
                                sanitized_value = value.replace("</p", "").replace("&#47;", "/")
                                sanitized_name = name.replace("-", "_").replace("/", "_")
                                sanitized_name = sanitized_name.replace("&#47;", "/")
                                if not sanitized_name.startswith("/"):
                                    server_status.status_dict[sanitized_name] = sanitized_value
                                break

            else:
                if response.error:
                    error = Exception(response.error)
                    lifecycle = ServerLifecycleState.ERROR
                elif response.response:
                    error = Exception(f"GET {self.config.status_url} request failed")
                    lifecycle = ServerLifecycleState.ERROR
                else:
                    error = Exception("unknown error")
                    lifecycle = ServerLifecycleState.UNKNOWN
                server_status.error = error

            server_status.at = lifecycle
            server_status.http_status_code = (response.response.status_code if response.response else None,)
            if server_status.at == ServerLifecycleState.READY:
                self.add_triple_count2_server_status(server_status)
        return server_status

    def test_geosparql(self) -> bool:
        """
        Test if GeoSPARQL functions work.

        Returns:
            True if GeoSPARQL is available
        """
        test_query = """
        PREFIX geo: <http://www.opengis.net/ont/geosparql#>
        PREFIX geof: <http://www.opengis.net/def/function/geosparql/>

        SELECT * WHERE {
            BIND(geof:distance("POINT(0 0)"^^geo:wktLiteral, "POINT(1 1)"^^geo:wktLiteral) AS ?dist)
        } LIMIT 1
        """

        response = self.make_request(
            "POST",
            self.sparql_url,
            data={"query": test_query},
            headers={"Accept": "application/sparql-results+json"},
        )

        geosparql_available = response.success
        return geosparql_available
__init__(config, env)

Initialize the Blazegraph manager.

Parameters:

Name Type Description Default
config ServerConfig

Server configuration

 required
env ServerEnv

Server environment (includes log, shell, debug, verbose)

 required
Source code in omnigraph/servers/blazegraph.py 
59
60
61
62
63
64
65
66
67
def __init__(self, config: ServerConfig, env: ServerEnv):
    """
    Initialize the Blazegraph manager.

    Args:
        config: Server configuration
        env: Server environment (includes log, shell, debug, verbose)
    """
    super().__init__(config=config, env=env)
get_dataloader_xml(container_path)

Build the DataLoader servlet properties XML for the given container-side file or directory path.

Parameters:

Name Type Description Default
container_path str

path as seen inside the container (mounted /data)

 required

Returns:

Type Description
str

the properties XML document
Source code in omnigraph/servers/blazegraph.py 
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
    def get_dataloader_xml(self, container_path: str) -> str:
        """
        Build the DataLoader servlet properties XML for the given
        container-side file or directory path.

        Args:
            container_path: path as seen inside the container (mounted /data)

        Returns:
            the properties XML document
        """
        xml = f"""<?xml version="1.0" encoding="UTF-8" standalone="no"?>
<!DOCTYPE properties SYSTEM "http://java.sun.com/dtd/properties.dtd">
<properties>
  <entry key="namespace">{self.config.dataset}</entry>
  <entry key="propertyFile">/RWStore.properties</entry>
  <entry key="fileOrDirs">{container_path}</entry>
  <entry key="-durableQueues">true</entry>
</properties>"""
        return xml
pre_create()

Prepare Blazegraph data directory and RWStore.properties.

Source code in omnigraph/servers/blazegraph.py 
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
def pre_create(self):
    """
    Prepare Blazegraph data directory and RWStore.properties.
    """
    data_dir = Path(self.config.base_data_dir)
    data_dir.mkdir(parents=True, exist_ok=True)
    rwstore_path = data_dir / "RWStore.properties"
    if not rwstore_path.exists():
        header = self.config.generator_header()
        props = f"""{header}

# Blazegraph journal configuration
com.bigdata.journal.AbstractJournal.file=/data/blazegraph.jnl

# Enable text index
com.bigdata.rdf.store.AbstractTripleStore.textIndex=true

# No OWL reasoning
com.bigdata.rdf.store.AbstractTripleStore.axiomsClass=com.bigdata.rdf.axioms.NoAxioms

# No justification or truth maintenance
com.bigdata.rdf.sail.truthMaintenance=false
com.bigdata.rdf.store.AbstractTripleStore.justify=false

# Default namespace
com.bigdata.rdf.sail.namespace={self.config.dataset}
"""
        rwstore_path.write_text(props)
status()

Get server status information.

Returns:

Type Description
ServerStatus

ServerStatus object with status information
Source code in omnigraph/servers/blazegraph.py 
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
def status(self) -> ServerStatus:
    """
    Get server status information.

    Returns:
        ServerStatus object with status information
    """
    server_status = super().status()
    if server_status.exists and server_status.running:
        response = self.make_request("GET", self.config.status_url)

        if response.success:
            lifecycle = ServerLifecycleState.READY

        if response.response and response.response.text:
            html_content = response.response.text
            # Only parse HTML if it looks like HTML content
            if "<" in html_content and ">" in html_content:
                name_value_pattern = r'(?:<span id="(?P<name1>[^"]+)">(?P<value1>[^<]+)</span[^>]*>|&#47;(?P<name2>[^=]+)=(?P<value2>[^\s&#]+))'
                matches = re.finditer(name_value_pattern, html_content, re.DOTALL)

                for match in matches:
                    for name_group, value_group in {
                        "name1": "value1",
                        "name2": "value2",
                    }.items():
                        name = match.group(name_group)
                        if name:
                            value = match.group(value_group)
                            sanitized_value = value.replace("</p", "").replace("&#47;", "/")
                            sanitized_name = name.replace("-", "_").replace("/", "_")
                            sanitized_name = sanitized_name.replace("&#47;", "/")
                            if not sanitized_name.startswith("/"):
                                server_status.status_dict[sanitized_name] = sanitized_value
                            break

        else:
            if response.error:
                error = Exception(response.error)
                lifecycle = ServerLifecycleState.ERROR
            elif response.response:
                error = Exception(f"GET {self.config.status_url} request failed")
                lifecycle = ServerLifecycleState.ERROR
            else:
                error = Exception("unknown error")
                lifecycle = ServerLifecycleState.UNKNOWN
            server_status.error = error

        server_status.at = lifecycle
        server_status.http_status_code = (response.response.status_code if response.response else None,)
        if server_status.at == ServerLifecycleState.READY:
            self.add_triple_count2_server_status(server_status)
    return server_status
test_geosparql()

Test if GeoSPARQL functions work.

Returns:

Type Description
bool

True if GeoSPARQL is available
Source code in omnigraph/servers/blazegraph.py 
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
def test_geosparql(self) -> bool:
    """
    Test if GeoSPARQL functions work.

    Returns:
        True if GeoSPARQL is available
    """
    test_query = """
    PREFIX geo: <http://www.opengis.net/ont/geosparql#>
    PREFIX geof: <http://www.opengis.net/def/function/geosparql/>

    SELECT * WHERE {
        BIND(geof:distance("POINT(0 0)"^^geo:wktLiteral, "POINT(1 1)"^^geo:wktLiteral) AS ?dist)
    } LIMIT 1
    """

    response = self.make_request(
        "POST",
        self.sparql_url,
        data={"query": test_query},
        headers={"Accept": "application/sparql-results+json"},
    )

    geosparql_available = response.success
    return geosparql_available
upload_dump_files(file_pattern=None)

Bulk-load dump files via Blazegraph's REST DataLoader servlet, which reads the files from the server's own filesystem (the mounted /data directory) instead of pushing them through HTTP request bodies.

Parameters:

Name Type Description Default
file_pattern str

Glob pattern for dump files

 None

Returns:

Type Description
int

Number of files loaded successfully
Source code in omnigraph/servers/blazegraph.py 
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
def upload_dump_files(self, file_pattern: str = None) -> int:
    """
    Bulk-load dump files via Blazegraph's REST DataLoader servlet, which
    reads the files from the server's own filesystem (the mounted /data
    directory) instead of pushing them through HTTP request bodies.

    Args:
        file_pattern: Glob pattern for dump files

    Returns:
        Number of files loaded successfully
    """
    container_name = self.config.container_name
    files = self.get_dump_files(file_pattern)
    loaded_count = 0
    if not files:
        self.log.log("⚠️", container_name, f"No dump files found for pattern: {file_pattern}")
    else:
        # stage the dumps under data_dir so the container sees them at /data/dumps
        stage_dir = Path(self.config.base_data_dir) / "dumps"
        stage_dir.mkdir(parents=True, exist_ok=True)
        for file in files:
            target = stage_dir / file.name
            if not target.exists():
                shutil.copy2(file, target)
        xml = self.get_dataloader_xml("/data/dumps")
        response = self.make_request(
            "POST",
            self.config.dataloader_url,
            headers={"Content-Type": "application/xml"},
            data=xml,
            timeout=self.config.upload_timeout,
        )
        if response.success:
            loaded_count = len(files)
            self.log.log("✅", container_name, f"DataLoader loaded {loaded_count} file(s)")
        else:
            error_msg = str(response.error) if response.error else f"HTTP {response.response.status_code}"
            self.log.log("❌", container_name, f"DataLoader failed: {error_msg}")
    return loaded_count

BlazegraphConfig dataclass

Bases: ServerConfig

Blazegraph configuration

Source code in omnigraph/servers/blazegraph.py 
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
@dataclass
class BlazegraphConfig(ServerConfig):
    """
    Blazegraph configuration
    """

    def __post_init__(self):
        super().__post_init__()
        blazegraph_base = f"{self.base_url}/bigdata"
        self.status_url = f"{blazegraph_base}/status"
        self.sparql_url = f"{blazegraph_base}/namespace/{self.dataset}/sparql"
        self.upload_url = self.sparql_url
        self.update_url = self.sparql_url
        self.web_url = f"{blazegraph_base}/#query"
        self.dataloader_url = f"{blazegraph_base}/dataloader"

    def get_docker_run_command(self, data_dir) -> str:
        """
        Generate docker run command with bind mount for Blazegraph journal directory.

        Args:
            data_dir: Host directory path to bind mount to container

        Returns:
            Complete docker run command string
        """
        docker_run_command = (
            f"docker run -d --name {self.container_name} "
            f"-e BLAZEGRAPH_UID={os.getuid()} "
            f"-e BLAZEGRAPH_GID={os.getgid()} "
            f"-p {self.docker_bind}:{self.port}:8080 "
            f"-v {data_dir}/RWStore.properties:/RWStore.properties "
            f"-v {data_dir}:/data "
            f"{self.image}"
        )
        return docker_run_command
get_docker_run_command(data_dir)

Generate docker run command with bind mount for Blazegraph journal directory.

Parameters:

Name Type Description Default
data_dir

Host directory path to bind mount to container

 required

Returns:

Type Description
str

Complete docker run command string
Source code in omnigraph/servers/blazegraph.py 
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
def get_docker_run_command(self, data_dir) -> str:
    """
    Generate docker run command with bind mount for Blazegraph journal directory.

    Args:
        data_dir: Host directory path to bind mount to container

    Returns:
        Complete docker run command string
    """
    docker_run_command = (
        f"docker run -d --name {self.container_name} "
        f"-e BLAZEGRAPH_UID={os.getuid()} "
        f"-e BLAZEGRAPH_GID={os.getgid()} "
        f"-p {self.docker_bind}:{self.port}:8080 "
        f"-v {data_dir}/RWStore.properties:/RWStore.properties "
        f"-v {data_dir}:/data "
        f"{self.image}"
    )
    return docker_run_command

graphdb

Created on 2025-05-30

Ontotext GraphDB SPARQL support

@author: wf

GraphDB

Bases: SparqlServer

Dockerized Ontotext GraphDB SPARQL server

Source code in omnigraph/servers/graphdb.py 
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
class GraphDB(SparqlServer):
    """
    Dockerized Ontotext GraphDB SPARQL server
    """

    def __init__(self, config: ServerConfig, env: ServerEnv):
        """
        Initialize the GraphDB manager.

        Args:
            config: Server configuration
            env: Server environment (includes log, shell, debug, verbose)
        """
        super().__init__(config=config, env=env)
        self.repo_created=False

    def post_start(self, first_start:bool):
        """Create repository after container starts.

        References:
            - https://graphdb.ontotext.com/documentation/11.2/manage-repos-with-restapi.html
        """
        if not first_start:
            return
        config = f"""@prefix rdfs: <http://www.w3.org/2000/01/rdf-schema#>.
    @prefix rep: <http://www.openrdf.org/config/repository#>.
    @prefix sr: <http://www.openrdf.org/config/repository/sail#>.
    @prefix sail: <http://www.openrdf.org/config/sail#>.
    @prefix graphdb: <http://www.ontotext.com/config/graphdb#>.

    [] a rep:Repository ;
        rep:repositoryID "{self.config.dataset}" ;
        rdfs:label "{self.config.dataset}" ;
        rep:repositoryImpl [
            rep:repositoryType "graphdb:SailRepository" ;
            sr:sailImpl [
                sail:sailType "graphdb:Sail" ;
                graphdb:read-only "false" ;
                graphdb:ruleset "rdfsplus-optimized" ;
                graphdb:disable-sameAs "true" ;
                graphdb:check-for-inconsistencies "false" ;
                graphdb:entity-id-size "32" ;
                graphdb:enable-context-index "false" ;
                graphdb:enablePredicateList "true" ;
                graphdb:enable-fts-index "false" ;
                graphdb:query-timeout "0" ;
                graphdb:throw-QueryEvaluationException-on-timeout "false" ;
                graphdb:query-limit-results "0" ;
                graphdb:base-URL "http://example.org/owlim#" ;
                graphdb:defaultNS "" ;
                graphdb:imports "" ;
                graphdb:repository-type "file-repository" ;
                graphdb:storage-folder "storage" ;
                graphdb:entity-index-size "10000000" ;
                graphdb:in-memory-literal-properties "true" ;
                graphdb:enable-literal-index "true" ;
            ]
        ] ."""

        files = {'config': ('repo-config.ttl', config, 'application/x-turtle')}

        response=self.make_request(
            "POST",
            f"{self.config.base_url}/rest/repositories",
            files=files
        )
        if not response.success:
            raise Exception(f"Failed to create repository: {response.error}")
        else:
            self.repo_created=True


    def status(self) -> ServerStatus:
        """
        Check GraphDB server status from container logs.

        Returns:
        ServerStatus object with status information
        """
        server_status = super().status()
        logs = server_status.logs
        if logs:
            if "Started GraphDB" in logs:
                lifecycle = ServerLifecycleState.READY
                server_status.at = lifecycle

        if server_status.at == ServerLifecycleState.READY:
            if self.repo_created:
                self.add_triple_count2_server_status(server_status)
        return server_status
__init__(config, env)

Initialize the GraphDB manager.

Parameters:

Name Type Description Default
config ServerConfig

Server configuration

 required
env ServerEnv

Server environment (includes log, shell, debug, verbose)

 required
Source code in omnigraph/servers/graphdb.py 
75
76
77
78
79
80
81
82
83
84
def __init__(self, config: ServerConfig, env: ServerEnv):
    """
    Initialize the GraphDB manager.

    Args:
        config: Server configuration
        env: Server environment (includes log, shell, debug, verbose)
    """
    super().__init__(config=config, env=env)
    self.repo_created=False
post_start(first_start)

Create repository after container starts.

References
  • https://graphdb.ontotext.com/documentation/11.2/manage-repos-with-restapi.html
Source code in omnigraph/servers/graphdb.py 
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
def post_start(self, first_start:bool):
    """Create repository after container starts.

    References:
        - https://graphdb.ontotext.com/documentation/11.2/manage-repos-with-restapi.html
    """
    if not first_start:
        return
    config = f"""@prefix rdfs: <http://www.w3.org/2000/01/rdf-schema#>.
@prefix rep: <http://www.openrdf.org/config/repository#>.
@prefix sr: <http://www.openrdf.org/config/repository/sail#>.
@prefix sail: <http://www.openrdf.org/config/sail#>.
@prefix graphdb: <http://www.ontotext.com/config/graphdb#>.

[] a rep:Repository ;
    rep:repositoryID "{self.config.dataset}" ;
    rdfs:label "{self.config.dataset}" ;
    rep:repositoryImpl [
        rep:repositoryType "graphdb:SailRepository" ;
        sr:sailImpl [
            sail:sailType "graphdb:Sail" ;
            graphdb:read-only "false" ;
            graphdb:ruleset "rdfsplus-optimized" ;
            graphdb:disable-sameAs "true" ;
            graphdb:check-for-inconsistencies "false" ;
            graphdb:entity-id-size "32" ;
            graphdb:enable-context-index "false" ;
            graphdb:enablePredicateList "true" ;
            graphdb:enable-fts-index "false" ;
            graphdb:query-timeout "0" ;
            graphdb:throw-QueryEvaluationException-on-timeout "false" ;
            graphdb:query-limit-results "0" ;
            graphdb:base-URL "http://example.org/owlim#" ;
            graphdb:defaultNS "" ;
            graphdb:imports "" ;
            graphdb:repository-type "file-repository" ;
            graphdb:storage-folder "storage" ;
            graphdb:entity-index-size "10000000" ;
            graphdb:in-memory-literal-properties "true" ;
            graphdb:enable-literal-index "true" ;
        ]
    ] ."""

    files = {'config': ('repo-config.ttl', config, 'application/x-turtle')}

    response=self.make_request(
        "POST",
        f"{self.config.base_url}/rest/repositories",
        files=files
    )
    if not response.success:
        raise Exception(f"Failed to create repository: {response.error}")
    else:
        self.repo_created=True
status()

Check GraphDB server status from container logs.

Returns: ServerStatus object with status information

Source code in omnigraph/servers/graphdb.py 
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
def status(self) -> ServerStatus:
    """
    Check GraphDB server status from container logs.

    Returns:
    ServerStatus object with status information
    """
    server_status = super().status()
    logs = server_status.logs
    if logs:
        if "Started GraphDB" in logs:
            lifecycle = ServerLifecycleState.READY
            server_status.at = lifecycle

    if server_status.at == ServerLifecycleState.READY:
        if self.repo_created:
            self.add_triple_count2_server_status(server_status)
    return server_status

GraphDBConfig dataclass

Bases: ServerConfig

GraphDB configuration

Source code in omnigraph/servers/graphdb.py 
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
@dataclass
class GraphDBConfig(ServerConfig):
    """
    GraphDB configuration
    """

    def __post_init__(self):
        """
        configure the configuration
        """
        super().__post_init__()

        # Clean URLs without credentials
        graphdb_repo = f"{self.base_url}/repositories/{self.dataset}"
        self.status_url = f"{self.base_url}/rest/info"
        self.sparql_url = f"{graphdb_repo}"
        self.update_url = f"{graphdb_repo}/statements"
        self.upload_url = f"{graphdb_repo}/statements"
        self.web_url = f"{self.base_url}/sparql"

    def get_docker_run_command(self, data_dir) -> str:
        """
        Generate docker run command with bind mount for data directory.
        Handles image swapping (effective_image) and conditional license injection.

        Args:
            data_dir: Host directory path to bind mount to container

        Returns:
            Complete docker run command string
        """
        # 1. Determine which image to use (Enterprise vs Free/Community)
        target_image = self.effective_image

        # 2. Build Environment Variables
        env_parts = []
        if self.auth_password:
            env_parts.append(f"-e GDB_JAVA_OPTS='-Dgraphdb.auth.token.secret={self.auth_password}'")

        # 3. Only inject the License Key if we are using the Enterprise image.
        # This prevents sending the license env var to the free/free-edition image if swapped.
        if target_image == self.image and self.license_env_var:
            env_parts.append(f"-e {self.license_env_var}")

        env_str = " " + " ".join(env_parts) if env_parts else ""

        # 4. Construct Command
        docker_run_command = (
            f"docker run {env_str} -d --name {self.container_name} "
            f"-p {self.docker_bind}:{self.port}:7200 "
            f"-v {data_dir}:/opt/graphdb/home "
            f"{target_image}"
        )
        return docker_run_command
__post_init__()

configure the configuration

Source code in omnigraph/servers/graphdb.py 
20
21
22
23
24
25
26
27
28
29
30
31
32
def __post_init__(self):
    """
    configure the configuration
    """
    super().__post_init__()

    # Clean URLs without credentials
    graphdb_repo = f"{self.base_url}/repositories/{self.dataset}"
    self.status_url = f"{self.base_url}/rest/info"
    self.sparql_url = f"{graphdb_repo}"
    self.update_url = f"{graphdb_repo}/statements"
    self.upload_url = f"{graphdb_repo}/statements"
    self.web_url = f"{self.base_url}/sparql"
get_docker_run_command(data_dir)

Generate docker run command with bind mount for data directory. Handles image swapping (effective_image) and conditional license injection.

Parameters:

Name Type Description Default
data_dir

Host directory path to bind mount to container

 required

Returns:

Type Description
str

Complete docker run command string
Source code in omnigraph/servers/graphdb.py 
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
def get_docker_run_command(self, data_dir) -> str:
    """
    Generate docker run command with bind mount for data directory.
    Handles image swapping (effective_image) and conditional license injection.

    Args:
        data_dir: Host directory path to bind mount to container

    Returns:
        Complete docker run command string
    """
    # 1. Determine which image to use (Enterprise vs Free/Community)
    target_image = self.effective_image

    # 2. Build Environment Variables
    env_parts = []
    if self.auth_password:
        env_parts.append(f"-e GDB_JAVA_OPTS='-Dgraphdb.auth.token.secret={self.auth_password}'")

    # 3. Only inject the License Key if we are using the Enterprise image.
    # This prevents sending the license env var to the free/free-edition image if swapped.
    if target_image == self.image and self.license_env_var:
        env_parts.append(f"-e {self.license_env_var}")

    env_str = " " + " ".join(env_parts) if env_parts else ""

    # 4. Construct Command
    docker_run_command = (
        f"docker run {env_str} -d --name {self.container_name} "
        f"-p {self.docker_bind}:{self.port}:7200 "
        f"-v {data_dir}:/opt/graphdb/home "
        f"{target_image}"
    )
    return docker_run_command

jena

Created on 2025-05-28

Apache Jena SPARQL support

@author: wf

Jena

Bases: SparqlServer

Dockerized Jena Fuseki SPARQL server

Source code in omnigraph/servers/jena.py 
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
class Jena(SparqlServer):
    """
    Dockerized Jena Fuseki SPARQL server
    """

    def __init__(self, config: ServerConfig, env: ServerEnv):
        """
        Initialize the Jena Fuseki manager.

        Args:
            config: Server configuration
            env: Server environment (includes log, shell, debug, verbose)
        """
        super().__init__(config=config, env=env)

    def status(self) -> ServerStatus:
        """
        Get server status information.

        Returns:
        ServerStatus object with status information
        """
        server_status = super().status()
        logs = server_status.logs
        if logs and "Creating dataset" in logs and "Fuseki is available :-)" in logs:
            server_status.at = ServerLifecycleState.READY

        return server_status

    def execute_update_query(self, update_query: str) -> tuple[any, Exception]:
        """
        Execute SPARQL UPDATE query using Jena's update endpoint.

        Jena Fuseki requires application/sparql-update content type for UPDATE operations.

        Args:
            update_query: SPARQL UPDATE query string

        Returns:
            Tuple of (response, exception)
        """
        result, error = self.execute_update_query_with_post(update_query)
        return result, error

    def get_tdbloader_command(self, files: List[Path]) -> str:
        """
        Build the tdb2.tdbloader docker command for the given dump files.

        The loader classes ship inside fuseki-server.jar, so the server's own
        image is reused with an overridden entrypoint. The TDB2 dataset is
        single-writer - the server must be stopped while the loader runs.

        Args:
            files: dump files to load (from the dumps directory)

        Returns:
            the docker run command string
        """
        loc = f"/fuseki/databases/{self.config.dataset}"
        dumps_dir = Path(self.config.dumps_dir)
        file_args = " ".join(f"/dumps/{file.name}" for file in files)
        command = (
            f"docker run --rm {self.config.docker_user_flag} --entrypoint java "
            f"-v {self.config.base_data_dir}:/fuseki "
            f"-v {dumps_dir}:/dumps "
            f"{self.config.image} "
            f"-cp /jena-fuseki/fuseki-server.jar tdb2.tdbloader "
            f"--loc {loc} {file_args}"
        )
        return command

    def upload_dump_files(self, file_pattern: str = None) -> int:
        """
        Bulk-load dump files with tdb2.tdbloader directly into the TDB2
        database files - the native path for the file-backed store, avoiding
        the HTTP single-POST transaction that faults the mmap'd node table
        (issue #25).

        Args:
            file_pattern: Glob pattern for dump files

        Returns:
            Number of files loaded successfully
        """
        container_name = self.config.container_name
        files = self.get_dump_files(file_pattern)
        loaded_count = 0
        if not files:
            self.log.log("⚠️", container_name, f"No dump files found for pattern: {file_pattern}")
        else:
            self.stop()
            loader_cmd = self.get_tdbloader_command(files)
            shell_result = self.run_shell_command(
                loader_cmd,
                success_msg=f"tdb2.tdbloader loaded {len(files)} file(s)",
                error_msg="tdb2.tdbloader failed",
            )
            if shell_result.success:
                loaded_count = len(files)
            self.start()
        return loaded_count
__init__(config, env)

Initialize the Jena Fuseki manager.

Parameters:

Name Type Description Default
config ServerConfig

Server configuration

 required
env ServerEnv

Server environment (includes log, shell, debug, verbose)

 required
Source code in omnigraph/servers/jena.py 
65
66
67
68
69
70
71
72
73
def __init__(self, config: ServerConfig, env: ServerEnv):
    """
    Initialize the Jena Fuseki manager.

    Args:
        config: Server configuration
        env: Server environment (includes log, shell, debug, verbose)
    """
    super().__init__(config=config, env=env)
execute_update_query(update_query)

Execute SPARQL UPDATE query using Jena's update endpoint.

Jena Fuseki requires application/sparql-update content type for UPDATE operations.

Parameters:

Name Type Description Default
update_query str

SPARQL UPDATE query string

 required

Returns:

Type Description
tuple[any, Exception]

Tuple of (response, exception)
Source code in omnigraph/servers/jena.py 
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
def execute_update_query(self, update_query: str) -> tuple[any, Exception]:
    """
    Execute SPARQL UPDATE query using Jena's update endpoint.

    Jena Fuseki requires application/sparql-update content type for UPDATE operations.

    Args:
        update_query: SPARQL UPDATE query string

    Returns:
        Tuple of (response, exception)
    """
    result, error = self.execute_update_query_with_post(update_query)
    return result, error
get_tdbloader_command(files)

Build the tdb2.tdbloader docker command for the given dump files.

The loader classes ship inside fuseki-server.jar, so the server's own image is reused with an overridden entrypoint. The TDB2 dataset is single-writer - the server must be stopped while the loader runs.

Parameters:

Name Type Description Default
files List[Path]

dump files to load (from the dumps directory)

 required

Returns:

Type Description
str

the docker run command string
Source code in omnigraph/servers/jena.py 
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
def get_tdbloader_command(self, files: List[Path]) -> str:
    """
    Build the tdb2.tdbloader docker command for the given dump files.

    The loader classes ship inside fuseki-server.jar, so the server's own
    image is reused with an overridden entrypoint. The TDB2 dataset is
    single-writer - the server must be stopped while the loader runs.

    Args:
        files: dump files to load (from the dumps directory)

    Returns:
        the docker run command string
    """
    loc = f"/fuseki/databases/{self.config.dataset}"
    dumps_dir = Path(self.config.dumps_dir)
    file_args = " ".join(f"/dumps/{file.name}" for file in files)
    command = (
        f"docker run --rm {self.config.docker_user_flag} --entrypoint java "
        f"-v {self.config.base_data_dir}:/fuseki "
        f"-v {dumps_dir}:/dumps "
        f"{self.config.image} "
        f"-cp /jena-fuseki/fuseki-server.jar tdb2.tdbloader "
        f"--loc {loc} {file_args}"
    )
    return command
status()

Get server status information.

Returns: ServerStatus object with status information

Source code in omnigraph/servers/jena.py 
75
76
77
78
79
80
81
82
83
84
85
86
87
def status(self) -> ServerStatus:
    """
    Get server status information.

    Returns:
    ServerStatus object with status information
    """
    server_status = super().status()
    logs = server_status.logs
    if logs and "Creating dataset" in logs and "Fuseki is available :-)" in logs:
        server_status.at = ServerLifecycleState.READY

    return server_status
upload_dump_files(file_pattern=None)

Bulk-load dump files with tdb2.tdbloader directly into the TDB2 database files - the native path for the file-backed store, avoiding the HTTP single-POST transaction that faults the mmap'd node table (issue #25).

Parameters:

Name Type Description Default
file_pattern str

Glob pattern for dump files

 None

Returns:

Type Description
int

Number of files loaded successfully
Source code in omnigraph/servers/jena.py 
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
def upload_dump_files(self, file_pattern: str = None) -> int:
    """
    Bulk-load dump files with tdb2.tdbloader directly into the TDB2
    database files - the native path for the file-backed store, avoiding
    the HTTP single-POST transaction that faults the mmap'd node table
    (issue #25).

    Args:
        file_pattern: Glob pattern for dump files

    Returns:
        Number of files loaded successfully
    """
    container_name = self.config.container_name
    files = self.get_dump_files(file_pattern)
    loaded_count = 0
    if not files:
        self.log.log("⚠️", container_name, f"No dump files found for pattern: {file_pattern}")
    else:
        self.stop()
        loader_cmd = self.get_tdbloader_command(files)
        shell_result = self.run_shell_command(
            loader_cmd,
            success_msg=f"tdb2.tdbloader loaded {len(files)} file(s)",
            error_msg="tdb2.tdbloader failed",
        )
        if shell_result.success:
            loaded_count = len(files)
        self.start()
    return loaded_count

JenaConfig dataclass

Bases: ServerConfig

Jena Fuseki configuration

Source code in omnigraph/servers/jena.py 
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
@dataclass
class JenaConfig(ServerConfig):
    """
    Jena Fuseki configuration
    """

    def __post_init__(self):
        """
        configure the configuration
        """
        super().__post_init__()

        # Clean URLs without credentials
        jena_base = f"{self.base_url}/ds"
        self.status_url = f"{self.base_url}/$/ping"
        self.sparql_url = f"{jena_base}/sparql"
        self.update_url = f"{jena_base}/update"
        self.upload_url = f"{jena_base}/data"
        self.web_url = f"{self.base_url}/#/dataset/ds/query"

    def get_docker_run_command(self, data_dir) -> str:
        """
        Generate docker run command with bind mount for data directory.

        Args:
            data_dir: Host directory path to bind mount to container

        Returns:
            Complete docker run command string
        """
        # Docker command setup
        env = "-e FUSEKI_DATASET_1=ds"
        if self.auth_password:
            env = f"{env} -e ADMIN_PASSWORD={self.auth_password}"
        docker_run_command = (
            f"docker run {self.docker_user_flag} {env} -d --name {self.container_name} "
            f"-p {self.docker_bind}:{self.port}:3030 "
            f"-v {data_dir}:/fuseki "
            f"{self.image}"
        )
        return docker_run_command
__post_init__()

configure the configuration

Source code in omnigraph/servers/jena.py 
23
24
25
26
27
28
29
30
31
32
33
34
35
def __post_init__(self):
    """
    configure the configuration
    """
    super().__post_init__()

    # Clean URLs without credentials
    jena_base = f"{self.base_url}/ds"
    self.status_url = f"{self.base_url}/$/ping"
    self.sparql_url = f"{jena_base}/sparql"
    self.update_url = f"{jena_base}/update"
    self.upload_url = f"{jena_base}/data"
    self.web_url = f"{self.base_url}/#/dataset/ds/query"
get_docker_run_command(data_dir)

Generate docker run command with bind mount for data directory.

Parameters:

Name Type Description Default
data_dir

Host directory path to bind mount to container

 required

Returns:

Type Description
str

Complete docker run command string
Source code in omnigraph/servers/jena.py 
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
def get_docker_run_command(self, data_dir) -> str:
    """
    Generate docker run command with bind mount for data directory.

    Args:
        data_dir: Host directory path to bind mount to container

    Returns:
        Complete docker run command string
    """
    # Docker command setup
    env = "-e FUSEKI_DATASET_1=ds"
    if self.auth_password:
        env = f"{env} -e ADMIN_PASSWORD={self.auth_password}"
    docker_run_command = (
        f"docker run {self.docker_user_flag} {env} -d --name {self.container_name} "
        f"-p {self.docker_bind}:{self.port}:3030 "
        f"-v {data_dir}:/fuseki "
        f"{self.image}"
    )
    return docker_run_command

millenniumdb

Created on 2025-11-25

MillenniumDB SPARQL support generated by Claude Sonnet 4.5 using

https://gitingest.com/ and the content of https://github.com/WolfgangFahl/pyomnigraph/issues/14

with the prompt

implement #14

@author: wf

MillenniumDB

Bases: SparqlServer

Dockerized MillenniumDB SPARQL server

MillenniumDB uses a two-step process: 1. Import data with mdb-import to create database 2. Run server with mdb-server pointing to created database

Source code in omnigraph/servers/millenniumdb.py 
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
class MillenniumDB(SparqlServer):
    """
    Dockerized MillenniumDB SPARQL server

    MillenniumDB uses a two-step process:
    1. Import data with mdb-import to create database
    2. Run server with mdb-server pointing to created database
    """

    def __init__(self, config: ServerConfig, env: ServerEnv):
        """
        Initialize the MillenniumDB manager.

        Args:
            config: Server configuration
            env: Server environment (includes log, shell, debug, verbose)
        """
        super().__init__(config=config, env=env)

    def pre_create(self):
        """
        Import data using mdb-import before starting the server.
        This is MillenniumDB's unique two-step setup process.
        """
        data_dir = Path(self.config.base_data_dir)
        data_dir.mkdir(parents=True, exist_ok=True)

        db_dir = data_dir / self.config.dataset

        # Check if database already exists
        catalog_file = db_dir / "catalog.dat"
        if catalog_file.exists():
            self.log.log(
                "✅",
                self.config.container_name,
                f"Database already exists at {db_dir}"
            )
            return

        # Find RDF files to import
        dumps_dir = Path(self.config.dumps_dir) if self.config.dumps_dir else data_dir
        rdf_files = list(dumps_dir.glob(f"*{self.rdf_format.extension}"))

        if not rdf_files:
            self.log.log(
                "⚠️",
                self.config.container_name,
                f"No RDF files found in {dumps_dir}"
            )
            # Create empty database anyway
            import_files = []
        else:
            import_files = [str(f) for f in rdf_files]

        # Run mdb-import to create database
        self.log.log(
            "✅",
            self.config.container_name,
            f"Importing {len(import_files)} files into database..."
        )

        # Build import command
        # Map both dumps_dir and data_dir to container
        files_arg = " ".join([f"/import/{Path(f).name}" for f in import_files])

        import_cmd = (
            f"docker run --rm "
            f"-v {dumps_dir}:/import "
            f"-v {data_dir}:/data "
            f"{self.config.image} "
            f"mdb-import {files_arg} /data/{self.config.dataset}"
        )

        import_result = self.shell.run(import_cmd, tee=self.verbose)

        if import_result.returncode != 0:
            self.log.log(
                "❌",
                self.config.container_name,
                f"Import failed: {import_result.stderr}"
            )
            raise RuntimeError(f"mdb-import failed: {import_result.stderr}")

        self.log.log(
            "✅",
            self.config.container_name,
            f"Database created at {db_dir}"
        )

    def status(self) -> ServerStatus:
        """
        Get server status information.

        Returns:
            ServerStatus object with status information
        """
        server_status = super().status()

        if server_status.exists and server_status.running:
            # Check if SPARQL endpoint is responding
            response = self.make_request(
                "POST",
                self.config.sparql_url,
                headers={"Content-Type": "application/sparql-query"},
                data="SELECT * WHERE { ?s ?p ?o } LIMIT 1"
            )

            if response.success:
                server_status.at = ServerLifecycleState.READY
                server_status.http_status_code = response.response.status_code
                self.add_triple_count2_server_status(server_status)
            elif response.error:
                server_status.at = ServerLifecycleState.ERROR
                server_status.error = response.error

        return server_status

    def upload_request(self, file_content: bytes) -> Response:
        """
        MillenniumDB doesn't support HTTP upload.
        Data must be imported using mdb-import before server starts.

        This method will raise an error if called.
        """
        error_msg = (
            "MillenniumDB does not support HTTP upload. "
            "Data must be imported using mdb-import before starting the server. "
            "Place RDF files in the dumps_dir and restart the server."
        )
        self.log.log("❌", self.config.container_name, error_msg)
        return Response(None, RuntimeError(error_msg))

    def load_dump_files(self, file_pattern: str = None) -> int:
        """
        Override load_dump_files to explain MillenniumDB's import process.

        For MillenniumDB, data must be imported during database creation,
        not after the server is running. This method will restart the server
        with a fresh import if called.
        """
        self.log.log(
            "ℹ️",
            self.config.container_name,
            "MillenniumDB requires data import before server start. Restarting with fresh import..."
        )

        # Stop and remove existing container
        self.stop()
        self.rm()

        # Remove existing database directory to force reimport
        db_dir = Path(self.config.base_data_dir) / self.config.dataset
        if db_dir.exists():
            import shutil
            shutil.rmtree(db_dir)
            self.log.log(
                "✅",
                self.config.container_name,
                f"Removed existing database at {db_dir}"
            )

        # Start will trigger pre_create which does the import
        started = self.start()

        if started:
            # Count triples to verify import
            count = self.count_triples()
            self.log.log(
                "✅",
                self.config.container_name,
                f"Database imported with {count:,} triples"
            )
            return 1  # Return 1 to indicate success
        else:
            return 0

    def get_web_url(self) -> str:
        """
        Return the MillenniumDB Web UI URL.
        """
        return self.config.web_url
__init__(config, env)

Initialize the MillenniumDB manager.

Parameters:

Name Type Description Default
config ServerConfig

Server configuration

 required
env ServerEnv

Server environment (includes log, shell, debug, verbose)

 required
Source code in omnigraph/servers/millenniumdb.py 
83
84
85
86
87
88
89
90
91
def __init__(self, config: ServerConfig, env: ServerEnv):
    """
    Initialize the MillenniumDB manager.

    Args:
        config: Server configuration
        env: Server environment (includes log, shell, debug, verbose)
    """
    super().__init__(config=config, env=env)
get_web_url()

Return the MillenniumDB Web UI URL.

Source code in omnigraph/servers/millenniumdb.py 
250
251
252
253
254
def get_web_url(self) -> str:
    """
    Return the MillenniumDB Web UI URL.
    """
    return self.config.web_url
load_dump_files(file_pattern=None)

Override load_dump_files to explain MillenniumDB's import process.

For MillenniumDB, data must be imported during database creation, not after the server is running. This method will restart the server with a fresh import if called.

Source code in omnigraph/servers/millenniumdb.py 
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
def load_dump_files(self, file_pattern: str = None) -> int:
    """
    Override load_dump_files to explain MillenniumDB's import process.

    For MillenniumDB, data must be imported during database creation,
    not after the server is running. This method will restart the server
    with a fresh import if called.
    """
    self.log.log(
        "ℹ️",
        self.config.container_name,
        "MillenniumDB requires data import before server start. Restarting with fresh import..."
    )

    # Stop and remove existing container
    self.stop()
    self.rm()

    # Remove existing database directory to force reimport
    db_dir = Path(self.config.base_data_dir) / self.config.dataset
    if db_dir.exists():
        import shutil
        shutil.rmtree(db_dir)
        self.log.log(
            "✅",
            self.config.container_name,
            f"Removed existing database at {db_dir}"
        )

    # Start will trigger pre_create which does the import
    started = self.start()

    if started:
        # Count triples to verify import
        count = self.count_triples()
        self.log.log(
            "✅",
            self.config.container_name,
            f"Database imported with {count:,} triples"
        )
        return 1  # Return 1 to indicate success
    else:
        return 0
pre_create()

Import data using mdb-import before starting the server. This is MillenniumDB's unique two-step setup process.

Source code in omnigraph/servers/millenniumdb.py 
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
def pre_create(self):
    """
    Import data using mdb-import before starting the server.
    This is MillenniumDB's unique two-step setup process.
    """
    data_dir = Path(self.config.base_data_dir)
    data_dir.mkdir(parents=True, exist_ok=True)

    db_dir = data_dir / self.config.dataset

    # Check if database already exists
    catalog_file = db_dir / "catalog.dat"
    if catalog_file.exists():
        self.log.log(
            "✅",
            self.config.container_name,
            f"Database already exists at {db_dir}"
        )
        return

    # Find RDF files to import
    dumps_dir = Path(self.config.dumps_dir) if self.config.dumps_dir else data_dir
    rdf_files = list(dumps_dir.glob(f"*{self.rdf_format.extension}"))

    if not rdf_files:
        self.log.log(
            "⚠️",
            self.config.container_name,
            f"No RDF files found in {dumps_dir}"
        )
        # Create empty database anyway
        import_files = []
    else:
        import_files = [str(f) for f in rdf_files]

    # Run mdb-import to create database
    self.log.log(
        "✅",
        self.config.container_name,
        f"Importing {len(import_files)} files into database..."
    )

    # Build import command
    # Map both dumps_dir and data_dir to container
    files_arg = " ".join([f"/import/{Path(f).name}" for f in import_files])

    import_cmd = (
        f"docker run --rm "
        f"-v {dumps_dir}:/import "
        f"-v {data_dir}:/data "
        f"{self.config.image} "
        f"mdb-import {files_arg} /data/{self.config.dataset}"
    )

    import_result = self.shell.run(import_cmd, tee=self.verbose)

    if import_result.returncode != 0:
        self.log.log(
            "❌",
            self.config.container_name,
            f"Import failed: {import_result.stderr}"
        )
        raise RuntimeError(f"mdb-import failed: {import_result.stderr}")

    self.log.log(
        "✅",
        self.config.container_name,
        f"Database created at {db_dir}"
    )
status()

Get server status information.

Returns:

Type Description
ServerStatus

ServerStatus object with status information
Source code in omnigraph/servers/millenniumdb.py 
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
def status(self) -> ServerStatus:
    """
    Get server status information.

    Returns:
        ServerStatus object with status information
    """
    server_status = super().status()

    if server_status.exists and server_status.running:
        # Check if SPARQL endpoint is responding
        response = self.make_request(
            "POST",
            self.config.sparql_url,
            headers={"Content-Type": "application/sparql-query"},
            data="SELECT * WHERE { ?s ?p ?o } LIMIT 1"
        )

        if response.success:
            server_status.at = ServerLifecycleState.READY
            server_status.http_status_code = response.response.status_code
            self.add_triple_count2_server_status(server_status)
        elif response.error:
            server_status.at = ServerLifecycleState.ERROR
            server_status.error = response.error

    return server_status
upload_request(file_content)

MillenniumDB doesn't support HTTP upload. Data must be imported using mdb-import before server starts.

This method will raise an error if called.

Source code in omnigraph/servers/millenniumdb.py 
191
192
193
194
195
196
197
198
199
200
201
202
203
204
def upload_request(self, file_content: bytes) -> Response:
    """
    MillenniumDB doesn't support HTTP upload.
    Data must be imported using mdb-import before server starts.

    This method will raise an error if called.
    """
    error_msg = (
        "MillenniumDB does not support HTTP upload. "
        "Data must be imported using mdb-import before starting the server. "
        "Place RDF files in the dumps_dir and restart the server."
    )
    self.log.log("❌", self.config.container_name, error_msg)
    return Response(None, RuntimeError(error_msg))

MillenniumDBConfig dataclass

Bases: ServerConfig

MillenniumDB configuration

Source code in omnigraph/servers/millenniumdb.py 
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
@dataclass
class MillenniumDBConfig(ServerConfig):
    """
    MillenniumDB configuration
    """

    def __post_init__(self):
        """
        Configure the MillenniumDB-specific URLs and settings
        """
        super().__post_init__()

        # MillenniumDB uses different ports for SPARQL and Web UI
        self.sparql_port = 1234
        self.web_port = 4321

        # Configure URLs
        self.status_url = f"http://{self.host}:{self.sparql_port}/sparql"
        self.sparql_url = f"http://{self.host}:{self.sparql_port}/sparql"
        self.upload_url = None  # MillenniumDB uses mdb-import, not HTTP upload
        self.web_url = f"http://{self.host}:{self.web_port}"

    def get_docker_run_command(self, data_dir: str) -> str:
        """
        Generate docker run command for MillenniumDB server.

        Note: This starts the server. Data must be imported first using mdb-import.

        Args:
            data_dir: Host directory path containing the imported database

        Returns:
            Complete docker run command string
        """
        # The database path inside the container will be /data/<dataset>
        db_path = f"/data/{self.dataset}"

        docker_run_command = (
            f"docker run -d --name {self.container_name} "
            f"-p {self.docker_bind}:{self.sparql_port}:1234 "
            f"-p {self.docker_bind}:{self.web_port}:4321 "
            f"-v {data_dir}:/data "
            f"{self.image} "
            f"mdb-server {db_path}"
        )
        return docker_run_command
__post_init__()

Configure the MillenniumDB-specific URLs and settings

Source code in omnigraph/servers/millenniumdb.py 
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
def __post_init__(self):
    """
    Configure the MillenniumDB-specific URLs and settings
    """
    super().__post_init__()

    # MillenniumDB uses different ports for SPARQL and Web UI
    self.sparql_port = 1234
    self.web_port = 4321

    # Configure URLs
    self.status_url = f"http://{self.host}:{self.sparql_port}/sparql"
    self.sparql_url = f"http://{self.host}:{self.sparql_port}/sparql"
    self.upload_url = None  # MillenniumDB uses mdb-import, not HTTP upload
    self.web_url = f"http://{self.host}:{self.web_port}"
get_docker_run_command(data_dir)

Generate docker run command for MillenniumDB server.

Note: This starts the server. Data must be imported first using mdb-import.

Parameters:

Name Type Description Default
data_dir str

Host directory path containing the imported database

 required

Returns:

Type Description
str

Complete docker run command string
Source code in omnigraph/servers/millenniumdb.py 
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
def get_docker_run_command(self, data_dir: str) -> str:
    """
    Generate docker run command for MillenniumDB server.

    Note: This starts the server. Data must be imported first using mdb-import.

    Args:
        data_dir: Host directory path containing the imported database

    Returns:
        Complete docker run command string
    """
    # The database path inside the container will be /data/<dataset>
    db_path = f"/data/{self.dataset}"

    docker_run_command = (
        f"docker run -d --name {self.container_name} "
        f"-p {self.docker_bind}:{self.sparql_port}:1234 "
        f"-p {self.docker_bind}:{self.web_port}:4321 "
        f"-v {data_dir}:/data "
        f"{self.image} "
        f"mdb-server {db_path}"
    )
    return docker_run_command

oxigraph

Created on 2025-06-03

Oxigraph SPARQL support https://github.com/oxigraph/oxigraph https://pyoxigraph.readthedocs.io/en/stable/

@author: wf

Oxigraph

Bases: SparqlServer

Dockerized Oxigraph SPARQL server

Source code in omnigraph/servers/oxigraph.py 
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
class Oxigraph(SparqlServer):
    """
    Dockerized Oxigraph SPARQL server
    """

    def __init__(self, config: ServerConfig, env: ServerEnv):
        """
        Initialize the Oxigraph manager.

        Args:
            config: Server configuration
            env: Server environment (includes log, shell, debug, verbose)
        """
        super().__init__(config=config, env=env)

    def status(self) -> ServerStatus:
        """
        Get server status information.

        Returns:
            ServerStatus object with status information
        """
        server_status = super().status()
        logs = server_status.logs

        if logs and (
            "Listening for requests at" in logs or "Oxigraph server started" in logs
        ):
            # Also try a lightweight HTTP request to confirm it's actually responding
            response = self.make_request("GET", self.config.status_url)
            if response.success:
                server_status.at = ServerLifecycleState.READY
                self.add_triple_count2_server_status(server_status)

        return server_status

    def execute_update_query(self, update_query: str) -> tuple[any, Exception]:
        """
        Execute SPARQL UPDATE query using Oxigraphs's update endpoint.

        Oxigraph requires application/sparql-update content type for UPDATE operations.
        see also how Jena does this

        Args:
            update_query: SPARQL UPDATE query string

        Returns:
            Tuple of (response, exception)
        """
        result, error = self.execute_update_query_with_post(update_query)
        return result, error
__init__(config, env)

Initialize the Oxigraph manager.

Parameters:

Name Type Description Default
config ServerConfig

Server configuration

 required
env ServerEnv

Server environment (includes log, shell, debug, verbose)

 required
Source code in omnigraph/servers/oxigraph.py 
62
63
64
65
66
67
68
69
70
def __init__(self, config: ServerConfig, env: ServerEnv):
    """
    Initialize the Oxigraph manager.

    Args:
        config: Server configuration
        env: Server environment (includes log, shell, debug, verbose)
    """
    super().__init__(config=config, env=env)
execute_update_query(update_query)

Execute SPARQL UPDATE query using Oxigraphs's update endpoint.

Oxigraph requires application/sparql-update content type for UPDATE operations. see also how Jena does this

Parameters:

Name Type Description Default
update_query str

SPARQL UPDATE query string

 required

Returns:

Type Description
tuple[any, Exception]

Tuple of (response, exception)
Source code in omnigraph/servers/oxigraph.py 
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
def execute_update_query(self, update_query: str) -> tuple[any, Exception]:
    """
    Execute SPARQL UPDATE query using Oxigraphs's update endpoint.

    Oxigraph requires application/sparql-update content type for UPDATE operations.
    see also how Jena does this

    Args:
        update_query: SPARQL UPDATE query string

    Returns:
        Tuple of (response, exception)
    """
    result, error = self.execute_update_query_with_post(update_query)
    return result, error
status()

Get server status information.

Returns:

Type Description
ServerStatus

ServerStatus object with status information
Source code in omnigraph/servers/oxigraph.py 
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
def status(self) -> ServerStatus:
    """
    Get server status information.

    Returns:
        ServerStatus object with status information
    """
    server_status = super().status()
    logs = server_status.logs

    if logs and (
        "Listening for requests at" in logs or "Oxigraph server started" in logs
    ):
        # Also try a lightweight HTTP request to confirm it's actually responding
        response = self.make_request("GET", self.config.status_url)
        if response.success:
            server_status.at = ServerLifecycleState.READY
            self.add_triple_count2_server_status(server_status)

    return server_status

OxigraphConfig dataclass

Bases: ServerConfig

Oxigraph configuration

Source code in omnigraph/servers/oxigraph.py 
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
@dataclass
class OxigraphConfig(ServerConfig):
    """
    Oxigraph configuration
    """

    def __post_init__(self):
        """
        configure the configuration
        """
        super().__post_init__()

        # Clean URLs without credentials
        self.status_url = f"{self.base_url}/"
        self.sparql_url = f"{self.base_url}/query"
        self.update_url = f"{self.base_url}/update"
        # Oxigraph follows SPARQL 1.1 HTTP RDF Update Protocol:
        # POST to /store creates a new named graph, use /store?default for default graph
        self.upload_url = f"{self.base_url}/store?default"
        self.web_url = f"{self.base_url}/"

    def get_docker_run_command(self, data_dir) -> str:
        """
        Generate docker run command with bind mount for data directory.

        Args:
            data_dir: Host directory path to bind mount to container

        Returns:
            Complete docker run command string
        """
        docker_run_command = (
            f"docker run {self.docker_user_flag} -d --name {self.container_name} "
            f"-p {self.docker_bind}:{self.port}:7878 "
            f"-v {data_dir}:/data "
            f"{self.image} serve --bind 0.0.0.0:7878 --location /data"
        )
        return docker_run_command
__post_init__()

configure the configuration

Source code in omnigraph/servers/oxigraph.py 
23
24
25
26
27
28
29
30
31
32
33
34
35
36
def __post_init__(self):
    """
    configure the configuration
    """
    super().__post_init__()

    # Clean URLs without credentials
    self.status_url = f"{self.base_url}/"
    self.sparql_url = f"{self.base_url}/query"
    self.update_url = f"{self.base_url}/update"
    # Oxigraph follows SPARQL 1.1 HTTP RDF Update Protocol:
    # POST to /store creates a new named graph, use /store?default for default graph
    self.upload_url = f"{self.base_url}/store?default"
    self.web_url = f"{self.base_url}/"
get_docker_run_command(data_dir)

Generate docker run command with bind mount for data directory.

Parameters:

Name Type Description Default
data_dir

Host directory path to bind mount to container

 required

Returns:

Type Description
str

Complete docker run command string
Source code in omnigraph/servers/oxigraph.py 
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
def get_docker_run_command(self, data_dir) -> str:
    """
    Generate docker run command with bind mount for data directory.

    Args:
        data_dir: Host directory path to bind mount to container

    Returns:
        Complete docker run command string
    """
    docker_run_command = (
        f"docker run {self.docker_user_flag} -d --name {self.container_name} "
        f"-p {self.docker_bind}:{self.port}:7878 "
        f"-v {data_dir}:/data "
        f"{self.image} serve --bind 0.0.0.0:7878 --location /data"
    )
    return docker_run_command

qlever

Created on 2025-05-28

@author: wf

QLever

Bases: SparqlServer

Dockerized QLever SPARQL server

Source code in omnigraph/servers/qlever.py 
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
class QLever(SparqlServer):
    """
    Dockerized QLever SPARQL server
    """

    def __init__(self, config: ServerConfig, env: ServerEnv):
        """
        Initialize the QLever server manager.

        Args:
            config: Server configuration
            env: Server environment (includes log, shell, debug, verbose)
        """
        super().__init__(config=config, env=env)

    def status(self) -> ServerStatus:
        """
        Check QLever server status from container logs.

        Returns:
        ServerStatus object with status information
        """
        server_status = super().status()

        # Treat UP as READY
        if server_status.at == ServerLifecycleState.UP:
            server_status.at = ServerLifecycleState.READY
        self.add_triple_count2_server_status(server_status)

        return server_status

    def get_step_list(self) -> List[Step]:
        step_list = [
            Step(
                name="setup-config",
                data_dir=self.data_dir,
                file_name="Qleverfile",
                setup_cmd=f"qlever setup-config {self.dataset}",
                step=1,
            ),
            Step(
                name="get-data",
                data_dir=self.data_dir,
                file_name=None,  # dynamically determined below
                setup_cmd=f"qlever get-data",
                step=2,
            ),
            Step(
                name="index",
                data_dir=self.data_dir,
                file_name=f"{self.dataset}.meta-data.json",
                setup_cmd=f"qlever index",
                step=3,
            ),
            Step(
                name="start",
                data_dir=self.data_dir,
                setup_cmd=f"qlever start --server-container {self.config.container_name}",
                step=4,
            ),
            Step(
                name="ui",
                data_dir=self.data_dir,
                setup_cmd=f"qlever ui",
                step=5,
            ),
        ]
        return step_list

    def handle_config(self, step: Step):
        """
        handle config setting for following step and
        patch QLeverfile to port
        """
        qlever_file = QLeverfile.ofFile(step.path)
        qlever_name = qlever_file.get("data", "NAME")
        self.config.access_token = qlever_file.get("server", "ACCESS_TOKEN")
        msg = f"qlever setup-config for {qlever_name} done"
        self.log.log("✅", self.config.container_name, msg)
        input_files = qlever_file.get("index", "input_files")
        # patch the port
        qlever_file.set("server", "port", str(self.config.port))
        # path the name
        qlever_file.save()
        # steps are index from 1 so step.step in the step_list
        # is the following step
        self.step_list[step.step].file_name = input_files

    def start(self, show_progress: bool = True) -> bool:
        """
        Start QLever using proper workflow.
        """
        if not self.config.data_dir:
            raise ValueError("Data directory needs to be specified")
        self.data_dir = self.config.data_dir
        if not os.path.exists(self.config.data_dir):
            raise ValueError(f"Data directory {self.data.dir} needs to exist")
        self.dataset = self.config.dataset
        started = False
        if self.dataset:
            steps = 0
            self.step_list = self.get_step_list()
            for step in self.step_list:
                step.perform(server=self)
                if not step.success:
                    break
                if step.name == "setup-config":
                    self.handle_config(step)
                steps = step.step

            if steps >= 5:
                started = self.wait_until_ready(show_progress=show_progress)

        return started

    def _get_access_token(self) -> Optional[str]:
        """
        Get the access token for QLever authentication.

        First checks if already set in config, otherwise reads from QLeverfile.

        Returns:
            Access token string or None if not found
        """
        # Return cached token if available
        if hasattr(self.config, "access_token") and self.config.access_token:
            return self.config.access_token

        # Try to read from QLeverfile
        if self.config.data_dir:
            qleverfile_path = Path(self.config.data_dir) / "Qleverfile"
            qlever_file = QLeverfile.ofFile(qleverfile_path)
            if qlever_file:
                access_token = qlever_file.get("server", "ACCESS_TOKEN")
                # Cache it for future use
                self.config.access_token = access_token
                return access_token

        return None

    def get_index_commands(self, files: List[Path]) -> List[str]:
        """
        Build the qlever CLI commands to (re)index the given dump files.

        QLever loads bulk data by building its index from files - there is no
        native HTTP bulk-write path; the SPARQL INSERT route is only suitable
        for small increments.

        Args:
            files: dump files staged in the data directory

        Returns:
            list of shell commands to run in the data directory
        """
        commands = [
            "qlever stop",
            "qlever index --overwrite-existing",
            f"qlever start --server-container {self.config.container_name}",
        ]
        return commands

    def upload_dump_files(self, file_pattern: str = None) -> int:
        """
        Bulk-load dump files by rebuilding the QLever index from them -
        the native path; the turtle-to-INSERT conversion of upload_request
        is unsuitable for bulk data.

        Args:
            file_pattern: Glob pattern for dump files

        Returns:
            Number of files loaded successfully
        """
        container_name = self.config.container_name
        files = self.get_dump_files(file_pattern)
        loaded_count = 0
        if not files:
            self.log.log("⚠️", container_name, f"No dump files found for pattern: {file_pattern}")
            return loaded_count
        data_dir = Path(self.config.data_dir)
        qleverfile_path = data_dir / "Qleverfile"
        qlever_file = QLeverfile.ofFile(qleverfile_path)
        if qlever_file is None:
            self.log.log("❌", container_name, f"no Qleverfile in {data_dir} - run start (setup-config) first")
            return loaded_count
        # stage the dumps in the data directory and register them as input files
        for file in files:
            target = data_dir / file.name
            if not target.exists():
                shutil.copy2(file, target)
        input_files = " ".join(file.name for file in files)
        qlever_file.set("index", "INPUT_FILES", input_files)
        qlever_file.save()
        ok = True
        for command in self.get_index_commands(files):
            shell_result = self.run_shell_command(f"cd {data_dir};{command}")
            if not shell_result.success:
                self.log.log("❌", container_name, f"failed: {command}")
                ok = False
                break
        if ok:
            loaded_count = len(files)
            self.log.log("✅", container_name, f"index rebuilt from {loaded_count} file(s)")
        return loaded_count

    def upload_request(self, file_content: bytes) -> Response:
        """Upload request for QLever using SPARQL INSERT statements."""
        turtle_data = file_content.decode("utf-8")
        sparql_insert = self._convert_turtle_to_insert(turtle_data)

        # Get access token - read from QLeverfile if not already set
        access_token = self._get_access_token()

        response = self.make_request(
            "POST",
            self.config.sparql_url,
            headers={
                "Content-Type": "application/sparql-update",
                "Authorization": f"Bearer {access_token}",
            },
            data=sparql_insert,
            timeout=self.config.upload_timeout,
        )
        return response

    def _convert_turtle_to_insert(self, turtle_data: str) -> str:
        """Convert Turtle data to SPARQL INSERT statement."""

        graph = rdflib.Graph()
        graph.parse(data=turtle_data, format="turtle")

        triples_list = []
        for subject, predicate, obj in graph:
            triple_str = f"{subject.n3()} {predicate.n3()} {obj.n3()} ."
            triples_list.append(triple_str)

        triples_block = "\n    ".join(triples_list)
        sparql_insert = f"INSERT DATA {{\n    {triples_block}\n}}"

        return sparql_insert
__init__(config, env)

Initialize the QLever server manager.

Parameters:

Name Type Description Default
config ServerConfig

Server configuration

 required
env ServerEnv

Server environment (includes log, shell, debug, verbose)

 required
Source code in omnigraph/servers/qlever.py 
107
108
109
110
111
112
113
114
115
def __init__(self, config: ServerConfig, env: ServerEnv):
    """
    Initialize the QLever server manager.

    Args:
        config: Server configuration
        env: Server environment (includes log, shell, debug, verbose)
    """
    super().__init__(config=config, env=env)
get_index_commands(files)

Build the qlever CLI commands to (re)index the given dump files.

QLever loads bulk data by building its index from files - there is no native HTTP bulk-write path; the SPARQL INSERT route is only suitable for small increments.

Parameters:

Name Type Description Default
files List[Path]

dump files staged in the data directory

 required

Returns:

Type Description
List[str]

list of shell commands to run in the data directory
Source code in omnigraph/servers/qlever.py 
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
def get_index_commands(self, files: List[Path]) -> List[str]:
    """
    Build the qlever CLI commands to (re)index the given dump files.

    QLever loads bulk data by building its index from files - there is no
    native HTTP bulk-write path; the SPARQL INSERT route is only suitable
    for small increments.

    Args:
        files: dump files staged in the data directory

    Returns:
        list of shell commands to run in the data directory
    """
    commands = [
        "qlever stop",
        "qlever index --overwrite-existing",
        f"qlever start --server-container {self.config.container_name}",
    ]
    return commands
handle_config(step)

handle config setting for following step and patch QLeverfile to port

Source code in omnigraph/servers/qlever.py 
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
def handle_config(self, step: Step):
    """
    handle config setting for following step and
    patch QLeverfile to port
    """
    qlever_file = QLeverfile.ofFile(step.path)
    qlever_name = qlever_file.get("data", "NAME")
    self.config.access_token = qlever_file.get("server", "ACCESS_TOKEN")
    msg = f"qlever setup-config for {qlever_name} done"
    self.log.log("✅", self.config.container_name, msg)
    input_files = qlever_file.get("index", "input_files")
    # patch the port
    qlever_file.set("server", "port", str(self.config.port))
    # path the name
    qlever_file.save()
    # steps are index from 1 so step.step in the step_list
    # is the following step
    self.step_list[step.step].file_name = input_files
start(show_progress=True)

Start QLever using proper workflow.

Source code in omnigraph/servers/qlever.py 
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
def start(self, show_progress: bool = True) -> bool:
    """
    Start QLever using proper workflow.
    """
    if not self.config.data_dir:
        raise ValueError("Data directory needs to be specified")
    self.data_dir = self.config.data_dir
    if not os.path.exists(self.config.data_dir):
        raise ValueError(f"Data directory {self.data.dir} needs to exist")
    self.dataset = self.config.dataset
    started = False
    if self.dataset:
        steps = 0
        self.step_list = self.get_step_list()
        for step in self.step_list:
            step.perform(server=self)
            if not step.success:
                break
            if step.name == "setup-config":
                self.handle_config(step)
            steps = step.step

        if steps >= 5:
            started = self.wait_until_ready(show_progress=show_progress)

    return started
status()

Check QLever server status from container logs.

Returns: ServerStatus object with status information

Source code in omnigraph/servers/qlever.py 
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
def status(self) -> ServerStatus:
    """
    Check QLever server status from container logs.

    Returns:
    ServerStatus object with status information
    """
    server_status = super().status()

    # Treat UP as READY
    if server_status.at == ServerLifecycleState.UP:
        server_status.at = ServerLifecycleState.READY
    self.add_triple_count2_server_status(server_status)

    return server_status
upload_dump_files(file_pattern=None)

Bulk-load dump files by rebuilding the QLever index from them - the native path; the turtle-to-INSERT conversion of upload_request is unsuitable for bulk data.

Parameters:

Name Type Description Default
file_pattern str

Glob pattern for dump files

 None

Returns:

Type Description
int

Number of files loaded successfully
Source code in omnigraph/servers/qlever.py 
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
def upload_dump_files(self, file_pattern: str = None) -> int:
    """
    Bulk-load dump files by rebuilding the QLever index from them -
    the native path; the turtle-to-INSERT conversion of upload_request
    is unsuitable for bulk data.

    Args:
        file_pattern: Glob pattern for dump files

    Returns:
        Number of files loaded successfully
    """
    container_name = self.config.container_name
    files = self.get_dump_files(file_pattern)
    loaded_count = 0
    if not files:
        self.log.log("⚠️", container_name, f"No dump files found for pattern: {file_pattern}")
        return loaded_count
    data_dir = Path(self.config.data_dir)
    qleverfile_path = data_dir / "Qleverfile"
    qlever_file = QLeverfile.ofFile(qleverfile_path)
    if qlever_file is None:
        self.log.log("❌", container_name, f"no Qleverfile in {data_dir} - run start (setup-config) first")
        return loaded_count
    # stage the dumps in the data directory and register them as input files
    for file in files:
        target = data_dir / file.name
        if not target.exists():
            shutil.copy2(file, target)
    input_files = " ".join(file.name for file in files)
    qlever_file.set("index", "INPUT_FILES", input_files)
    qlever_file.save()
    ok = True
    for command in self.get_index_commands(files):
        shell_result = self.run_shell_command(f"cd {data_dir};{command}")
        if not shell_result.success:
            self.log.log("❌", container_name, f"failed: {command}")
            ok = False
            break
    if ok:
        loaded_count = len(files)
        self.log.log("✅", container_name, f"index rebuilt from {loaded_count} file(s)")
    return loaded_count
upload_request(file_content)

Upload request for QLever using SPARQL INSERT statements.

Source code in omnigraph/servers/qlever.py 
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
def upload_request(self, file_content: bytes) -> Response:
    """Upload request for QLever using SPARQL INSERT statements."""
    turtle_data = file_content.decode("utf-8")
    sparql_insert = self._convert_turtle_to_insert(turtle_data)

    # Get access token - read from QLeverfile if not already set
    access_token = self._get_access_token()

    response = self.make_request(
        "POST",
        self.config.sparql_url,
        headers={
            "Content-Type": "application/sparql-update",
            "Authorization": f"Bearer {access_token}",
        },
        data=sparql_insert,
        timeout=self.config.upload_timeout,
    )
    return response

QLeverConfig dataclass

Bases: ServerConfig

specialized QLever configuration

Source code in omnigraph/servers/qlever.py 
86
87
88
89
90
91
92
93
94
95
96
97
98
@dataclass
class QLeverConfig(ServerConfig):
    """
    specialized QLever configuration
    """

    def __post_init__(self):
        super().__post_init__()
        self.access_token = None
        self.status_url = f"{self.base_url}"
        self.sparql_url = f"{self.base_url}/api/sparql"
        # the docker run command is dynamically created by the qlever (control) command later
        self.docker_run_command = None

QLeverfile

handle qlever control https://github.com/ad-freiburg/qlever-control QLeverfile in INI format

Source code in omnigraph/servers/qlever.py 
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
class QLeverfile:
    """
    handle qlever control https://github.com/ad-freiburg/qlever-control
    QLeverfile in INI format
    """

    def __init__(self, path: Path, config: ConfigParser):
        self.path = path
        self.config = config

    @classmethod
    def ofFile(cls, path: Path) -> Optional["QLeverfile"]:
        """
        Create QLeverfile instance from given INI file
        """
        if not path.exists():
            return None
        config = ConfigParser(interpolation=ExtendedInterpolation())
        config.read(path)
        return cls(path, config)

    def get(self, section: str, key: str) -> Optional[str]:
        """
        Get a value from the config, if exists
        """
        if self.config.has_section(section) and self.config.has_option(section, key):
            return self.config.get(section, key)
        return None

    def set(self, section: str, key: str, value: str):
        """
        Set a value in the config
        """
        if not self.config.has_section(section):
            self.config.add_section(section)
        self.config.set(section, key, value)

    def sections(self) -> list[str]:
        """
        Return list of config sections
        """
        return self.config.sections()

    def as_dict(self) -> dict[str, dict[str, str]]:
        """
        Return full config as nested dictionary
        """
        return {
            section: dict(self.config.items(section))
            for section in self.config.sections()
        }

    def save(self):
        """
        Save the config back to file
        """
        with self.path.open("w") as f:
            self.config.write(f)
as_dict()

Return full config as nested dictionary

Source code in omnigraph/servers/qlever.py 
69
70
71
72
73
74
75
76
def as_dict(self) -> dict[str, dict[str, str]]:
    """
    Return full config as nested dictionary
    """
    return {
        section: dict(self.config.items(section))
        for section in self.config.sections()
    }
get(section, key)

Get a value from the config, if exists

Source code in omnigraph/servers/qlever.py 
47
48
49
50
51
52
53
def get(self, section: str, key: str) -> Optional[str]:
    """
    Get a value from the config, if exists
    """
    if self.config.has_section(section) and self.config.has_option(section, key):
        return self.config.get(section, key)
    return None
ofFile(path) classmethod

Create QLeverfile instance from given INI file

Source code in omnigraph/servers/qlever.py 
36
37
38
39
40
41
42
43
44
45
@classmethod
def ofFile(cls, path: Path) -> Optional["QLeverfile"]:
    """
    Create QLeverfile instance from given INI file
    """
    if not path.exists():
        return None
    config = ConfigParser(interpolation=ExtendedInterpolation())
    config.read(path)
    return cls(path, config)
save()

Save the config back to file

Source code in omnigraph/servers/qlever.py 
78
79
80
81
82
83
def save(self):
    """
    Save the config back to file
    """
    with self.path.open("w") as f:
        self.config.write(f)
sections()

Return list of config sections

Source code in omnigraph/servers/qlever.py 
63
64
65
66
67
def sections(self) -> list[str]:
    """
    Return list of config sections
    """
    return self.config.sections()
set(section, key, value)

Set a value in the config

Source code in omnigraph/servers/qlever.py 
55
56
57
58
59
60
61
def set(self, section: str, key: str, value: str):
    """
    Set a value in the config
    """
    if not self.config.has_section(section):
        self.config.add_section(section)
    self.config.set(section, key, value)

stardog

Created on 2025-06-03

Stardog SPARQL support

@author: wf

Stardog

Bases: SparqlServer

Dockerized Stardog SPARQL server

Source code in omnigraph/servers/stardog.py 
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
class Stardog(SparqlServer):
    """
    Dockerized Stardog SPARQL server
    """

    def __init__(self, config: ServerConfig, env: ServerEnv):
        """
        Initialize the Stardog manager.

        Args:
            config: Server configuration
            env: Server environment (includes log, shell, debug, verbose)
        """
        super().__init__(config=config, env=env)

    def pre_create(self):
        """Build Stardog image if needed."""
        if not self.config.has_license:
            return

        target = self.effective_image

        # Check if image exists
        result = self.shell.run(f"docker images -q {target}")
        if result.stdout.strip():
            self.msg(f"Image {target} already exists")
            return

        # Build image
        self.msg(f"Building {target}...")
        dockerfile_path = self.config_path / "Dockerfile"
        result = self.shell.run(
            f"docker build -t {target} -f {dockerfile_path} {self.config_path}",
            tee=True
        )
        if result.returncode == 0:
            self.msg(f"✅ Built {target}")
        else:
            raise RuntimeError(f"Failed to build {target}")

    def status(self) -> ServerStatus:
        """
        Get server status information.

        Returns:
            ServerStatus object with status information
        """
        server_status = super().status()
        logs = server_status.logs

        if logs and "Stardog server started" in logs and "Server is ready" in logs:
            server_status.at = ServerLifecycleState.READY
        return server_status
__init__(config, env)

Initialize the Stardog manager.

Parameters:

Name Type Description Default
config ServerConfig

Server configuration

 required
env ServerEnv

Server environment (includes log, shell, debug, verbose)

 required
Source code in omnigraph/servers/stardog.py 
77
78
79
80
81
82
83
84
85
def __init__(self, config: ServerConfig, env: ServerEnv):
    """
    Initialize the Stardog manager.

    Args:
        config: Server configuration
        env: Server environment (includes log, shell, debug, verbose)
    """
    super().__init__(config=config, env=env)
pre_create()

Build Stardog image if needed.

Source code in omnigraph/servers/stardog.py 
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
def pre_create(self):
    """Build Stardog image if needed."""
    if not self.config.has_license:
        return

    target = self.effective_image

    # Check if image exists
    result = self.shell.run(f"docker images -q {target}")
    if result.stdout.strip():
        self.msg(f"Image {target} already exists")
        return

    # Build image
    self.msg(f"Building {target}...")
    dockerfile_path = self.config_path / "Dockerfile"
    result = self.shell.run(
        f"docker build -t {target} -f {dockerfile_path} {self.config_path}",
        tee=True
    )
    if result.returncode == 0:
        self.msg(f"✅ Built {target}")
    else:
        raise RuntimeError(f"Failed to build {target}")
status()

Get server status information.

Returns:

Type Description
ServerStatus

ServerStatus object with status information
Source code in omnigraph/servers/stardog.py 
112
113
114
115
116
117
118
119
120
121
122
123
124
def status(self) -> ServerStatus:
    """
    Get server status information.

    Returns:
        ServerStatus object with status information
    """
    server_status = super().status()
    logs = server_status.logs

    if logs and "Stardog server started" in logs and "Server is ready" in logs:
        server_status.at = ServerLifecycleState.READY
    return server_status

StardogConfig dataclass

Bases: ServerConfig

Stardog configuration

Source code in omnigraph/servers/stardog.py 
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
@dataclass
class StardogConfig(ServerConfig):
    """
    Stardog configuration
    """

    def __post_init__(self):
        """
        configure the configuration
        """
        super().__post_init__()

        # Clean URLs without credentials
        stardog_base = f"{self.base_url}/{self.dataset}"
        self.status_url = f"{self.base_url}/admin/status"
        self.sparql_url = f"{stardog_base}/query"
        self.update_url = f"{stardog_base}/update"
        self.upload_url = f"{stardog_base}/add"
        self.web_url = f"{self.base_url}/"

    def get_docker_run_command(self, data_dir) -> str:
        """
        Generate docker run command with bind mount for data directory.
        Handles the special case of swapping to the free image (Github/Zazuko)
        where the license key must NOT be injected.

        Args:
            data_dir: Host directory path to bind mount to container

        Returns:
            Complete docker run command string
        """
        # 1. Determine which image to use (Enterprise vs Free/Github) based on license existence
        target_image = self.effective_image

        # 2. Build Environment Variables
        env_parts = []
        if self.auth_password:
            env_parts.append("-e STARDOG_SERVER_JAVA_ARGS='-Dstardog.default.cli.server=http://localhost:5820'")

        # 3. Only inject the License Key if we are using the Enterprise image.
        # The free (Github) image does not require (and may not support) the license env var.
        if target_image == self.image and self.license_env_var:
            env_parts.append(f"-e {self.license_env_var}")

        env_str = " " + " ".join(env_parts) if env_parts else ""

        # 4. Construct Command using target_image
        docker_run_command = (
            f"docker run {self.docker_user_flag}{env_str} -d --name {self.container_name} "
            f"-p {self.docker_bind}:{self.port}:5820 "
            f"-v {data_dir}:/var/opt/stardog "
            f"{target_image}"
        )
        return docker_run_command
__post_init__()

configure the configuration

Source code in omnigraph/servers/stardog.py 
21
22
23
24
25
26
27
28
29
30
31
32
33
def __post_init__(self):
    """
    configure the configuration
    """
    super().__post_init__()

    # Clean URLs without credentials
    stardog_base = f"{self.base_url}/{self.dataset}"
    self.status_url = f"{self.base_url}/admin/status"
    self.sparql_url = f"{stardog_base}/query"
    self.update_url = f"{stardog_base}/update"
    self.upload_url = f"{stardog_base}/add"
    self.web_url = f"{self.base_url}/"
get_docker_run_command(data_dir)

Generate docker run command with bind mount for data directory. Handles the special case of swapping to the free image (Github/Zazuko) where the license key must NOT be injected.

Parameters:

Name Type Description Default
data_dir

Host directory path to bind mount to container

 required

Returns:

Type Description
str

Complete docker run command string
Source code in omnigraph/servers/stardog.py 
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
def get_docker_run_command(self, data_dir) -> str:
    """
    Generate docker run command with bind mount for data directory.
    Handles the special case of swapping to the free image (Github/Zazuko)
    where the license key must NOT be injected.

    Args:
        data_dir: Host directory path to bind mount to container

    Returns:
        Complete docker run command string
    """
    # 1. Determine which image to use (Enterprise vs Free/Github) based on license existence
    target_image = self.effective_image

    # 2. Build Environment Variables
    env_parts = []
    if self.auth_password:
        env_parts.append("-e STARDOG_SERVER_JAVA_ARGS='-Dstardog.default.cli.server=http://localhost:5820'")

    # 3. Only inject the License Key if we are using the Enterprise image.
    # The free (Github) image does not require (and may not support) the license env var.
    if target_image == self.image and self.license_env_var:
        env_parts.append(f"-e {self.license_env_var}")

    env_str = " " + " ".join(env_parts) if env_parts else ""

    # 4. Construct Command using target_image
    docker_run_command = (
        f"docker run {self.docker_user_flag}{env_str} -d --name {self.container_name} "
        f"-p {self.docker_bind}:{self.port}:5820 "
        f"-v {data_dir}:/var/opt/stardog "
        f"{target_image}"
    )
    return docker_run_command

virtuoso

Created on 2025-06-03

OpenLink Virtuoso SPARQL support

@author: wf

Virtuoso

Bases: SparqlServer

Dockerized OpenLink Virtuoso SPARQL server

Source code in omnigraph/servers/virtuoso.py 
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
class Virtuoso(SparqlServer):
    """
    Dockerized OpenLink Virtuoso SPARQL server
    """

    def __init__(self, config: ServerConfig, env: ServerEnv):
        """
        Initialize the Virtuoso manager.

        Args:
            config: Server configuration
            env: Server environment (includes log, shell, debug, verbose)
        """
        super().__init__(config=config, env=env)

    def post_create(self):
        """
        Setup permissions after container creation.
        """
        super().post_create()
        self.setup_permissions()

    def run_isql_cmd(self, cmd: str) -> ShellResult:
        """
        Run SQL command via isql.
        """
        # Escape double quotes in the SQL command for proper shell handling
        escaped_cmd = cmd.replace('"', '\\"')
        args = (
            f'isql 1111 dba {self.config.auth_password or "dba"} "EXEC={escaped_cmd}"'
        )
        shell_result = self.run_docker_cmd("exec", args=args)
        return shell_result

    def setup_permissions(self) -> bool:
        """
        Grant necessary permissions to SPARQL user.
        """
        # Grant general SPARQL update capability
        success = True
        grants = [
            'GRANT SPARQL_UPDATE TO "SPARQL";',
            # workaround of 2023-01 as per https://community.openlinksw.com/t/sparul-insert-access-denied-even-after-granting-update-permission/3448/7
            "DB.DBA.RDF_DEFAULT_USER_PERMS_SET ('nobody', 7);",
            # Grant write permissions on default graph for SPARQL user
            "DB.DBA.RDF_DEFAULT_USER_PERMS_SET ('SPARQL', 7);",
        ]
        for sql in grants:
            shell_result = self.run_isql_cmd(sql)
            success = success and shell_result.success

        return success

    def status(self) -> ServerStatus:
        """
        Get server status information.

        Returns:
            ServerStatus object with status information
        """
        server_status = super().status()
        logs = server_status.logs

        if (
            logs
            and "Server online at" in logs
            and "HTTP/WebDAV server online at" in logs
        ):
            server_status.at = ServerLifecycleState.READY

        return server_status

    def ensure_permissions(self):
        """
        Ensure permissions are set (can be called even if server is already running).
        """
        status = self.status()
        if status.running:
            self.setup_permissions()

    def get_clear_query(self) -> str:
        """
        the clear query to be used
        overrides the default query
        """
        # Ensure permissions are set before clearing
        self.ensure_permissions()

        # Use CLEAR GRAPH instead of DELETE for better Virtuoso compatibility
        # This requires fewer permissions than DELETE
        clear_query = "CLEAR GRAPH <urn:virtuoso:default>"
        return clear_query

    def get_web_url(self) -> str:
        web_url = self.config.web_url
        if self.config.auth_user and self.config.auth_password:
            proto, rest = web_url.split("://", 1)
            auth = f"{self.config.auth_user}:{self.config.auth_password}@"
            web_url = f"{proto}://{auth}{rest}"
        return web_url
__init__(config, env)

Initialize the Virtuoso manager.

Parameters:

Name Type Description Default
config ServerConfig

Server configuration

 required
env ServerEnv

Server environment (includes log, shell, debug, verbose)

 required
Source code in omnigraph/servers/virtuoso.py 
64
65
66
67
68
69
70
71
72
def __init__(self, config: ServerConfig, env: ServerEnv):
    """
    Initialize the Virtuoso manager.

    Args:
        config: Server configuration
        env: Server environment (includes log, shell, debug, verbose)
    """
    super().__init__(config=config, env=env)
ensure_permissions()

Ensure permissions are set (can be called even if server is already running).

Source code in omnigraph/servers/virtuoso.py 
131
132
133
134
135
136
137
def ensure_permissions(self):
    """
    Ensure permissions are set (can be called even if server is already running).
    """
    status = self.status()
    if status.running:
        self.setup_permissions()
get_clear_query()

the clear query to be used overrides the default query

Source code in omnigraph/servers/virtuoso.py 
139
140
141
142
143
144
145
146
147
148
149
150
def get_clear_query(self) -> str:
    """
    the clear query to be used
    overrides the default query
    """
    # Ensure permissions are set before clearing
    self.ensure_permissions()

    # Use CLEAR GRAPH instead of DELETE for better Virtuoso compatibility
    # This requires fewer permissions than DELETE
    clear_query = "CLEAR GRAPH <urn:virtuoso:default>"
    return clear_query
post_create()

Setup permissions after container creation.

Source code in omnigraph/servers/virtuoso.py 
74
75
76
77
78
79
def post_create(self):
    """
    Setup permissions after container creation.
    """
    super().post_create()
    self.setup_permissions()
run_isql_cmd(cmd)

Run SQL command via isql.

Source code in omnigraph/servers/virtuoso.py 
81
82
83
84
85
86
87
88
89
90
91
def run_isql_cmd(self, cmd: str) -> ShellResult:
    """
    Run SQL command via isql.
    """
    # Escape double quotes in the SQL command for proper shell handling
    escaped_cmd = cmd.replace('"', '\\"')
    args = (
        f'isql 1111 dba {self.config.auth_password or "dba"} "EXEC={escaped_cmd}"'
    )
    shell_result = self.run_docker_cmd("exec", args=args)
    return shell_result
setup_permissions()

Grant necessary permissions to SPARQL user.

Source code in omnigraph/servers/virtuoso.py 
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
def setup_permissions(self) -> bool:
    """
    Grant necessary permissions to SPARQL user.
    """
    # Grant general SPARQL update capability
    success = True
    grants = [
        'GRANT SPARQL_UPDATE TO "SPARQL";',
        # workaround of 2023-01 as per https://community.openlinksw.com/t/sparul-insert-access-denied-even-after-granting-update-permission/3448/7
        "DB.DBA.RDF_DEFAULT_USER_PERMS_SET ('nobody', 7);",
        # Grant write permissions on default graph for SPARQL user
        "DB.DBA.RDF_DEFAULT_USER_PERMS_SET ('SPARQL', 7);",
    ]
    for sql in grants:
        shell_result = self.run_isql_cmd(sql)
        success = success and shell_result.success

    return success
status()

Get server status information.

Returns:

Type Description
ServerStatus

ServerStatus object with status information
Source code in omnigraph/servers/virtuoso.py 
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
def status(self) -> ServerStatus:
    """
    Get server status information.

    Returns:
        ServerStatus object with status information
    """
    server_status = super().status()
    logs = server_status.logs

    if (
        logs
        and "Server online at" in logs
        and "HTTP/WebDAV server online at" in logs
    ):
        server_status.at = ServerLifecycleState.READY

    return server_status

VirtuosoConfig dataclass

Bases: ServerConfig

Virtuoso configuration

Source code in omnigraph/servers/virtuoso.py 
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
@dataclass
class VirtuosoConfig(ServerConfig):
    """
    Virtuoso configuration
    """

    def __post_init__(self):
        """
        configure the configuration
        """
        super().__post_init__()

        # Clean URLs without credentials
        self.status_url = f"{self.base_url}/sparql"
        self.sparql_url = f"{self.base_url}/sparql"
        self.update_url = f"{self.base_url}/sparql"
        self.upload_url = f"{self.base_url}/sparql-graph-crud"
        self.web_url = f"{self.base_url}/sparql"

    def get_docker_run_command(self, data_dir) -> str:
        """
        Generate docker run command with bind mount for data directory.

        Args:
            data_dir: Host directory path to bind mount to container

        Returns:
            Complete docker run command string
        """
        # Docker command setup
        env = "-e SPARQL_UPDATE=true"
        if self.auth_password:
            env += f" -e DBA_PASSWORD={self.auth_password}"

        # run as root - no user flag
        docker_run_command = (
            f"docker run {env} -d --name {self.container_name} "
            f"-p {self.docker_bind}:{self.port}:8890 "
            f"-v {data_dir}:/database "
            f"{self.image}"
        )
        return docker_run_command
__post_init__()

configure the configuration

Source code in omnigraph/servers/virtuoso.py 
21
22
23
24
25
26
27
28
29
30
31
32
def __post_init__(self):
    """
    configure the configuration
    """
    super().__post_init__()

    # Clean URLs without credentials
    self.status_url = f"{self.base_url}/sparql"
    self.sparql_url = f"{self.base_url}/sparql"
    self.update_url = f"{self.base_url}/sparql"
    self.upload_url = f"{self.base_url}/sparql-graph-crud"
    self.web_url = f"{self.base_url}/sparql"
get_docker_run_command(data_dir)

Generate docker run command with bind mount for data directory.

Parameters:

Name Type Description Default
data_dir

Host directory path to bind mount to container

 required

Returns:

Type Description
str

Complete docker run command string
Source code in omnigraph/servers/virtuoso.py 
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
def get_docker_run_command(self, data_dir) -> str:
    """
    Generate docker run command with bind mount for data directory.

    Args:
        data_dir: Host directory path to bind mount to container

    Returns:
        Complete docker run command string
    """
    # Docker command setup
    env = "-e SPARQL_UPDATE=true"
    if self.auth_password:
        env += f" -e DBA_PASSWORD={self.auth_password}"

    # run as root - no user flag
    docker_run_command = (
        f"docker run {env} -d --name {self.container_name} "
        f"-p {self.docker_bind}:{self.port}:8890 "
        f"-v {data_dir}:/database "
        f"{self.image}"
    )
    return docker_run_command

software

Software dataclass

Single software requirement definition

Source code in omnigraph/software.py 
10
11
12
13
14
15
16
17
@dataclass
class Software:
    """
    Single software requirement definition
    """

    command: str
    info: str

SoftwareList

Collection of software requirements loadable from YAML

Source code in omnigraph/software.py 
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
@lod_storable
class SoftwareList:
    """
    Collection of software requirements loadable from YAML
    """

    software_list: List[Software] = field(default_factory=list)

    def check_installed(self, log: Log, shell: Shell, verbose: bool = True) -> int:
        """
        Check if necessary software
        commands are available and suggest installation packages
        """
        missing_counter = 0

        log.log("✅", "info", f"PATH={os.environ.get('PATH')}")

        for needed in self.software_list:
            process = shell.run(f"which {needed.command}", tee=verbose)
            where = None if process.returncode != 0 else process.stdout.strip()

            if not where:
                log.log("❌", "error", f"Missing required command: {needed.command} - {needed.info}")
                missing_counter += 1
            else:
                log.log("✅", "info", f"{needed.command} available at {where}")

        return missing_counter

check_installed(log, shell, verbose=True)

Check if necessary software commands are available and suggest installation packages

Source code in omnigraph/software.py 
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
def check_installed(self, log: Log, shell: Shell, verbose: bool = True) -> int:
    """
    Check if necessary software
    commands are available and suggest installation packages
    """
    missing_counter = 0

    log.log("✅", "info", f"PATH={os.environ.get('PATH')}")

    for needed in self.software_list:
        process = shell.run(f"which {needed.command}", tee=verbose)
        where = None if process.returncode != 0 else process.stdout.strip()

        if not where:
            log.log("❌", "error", f"Missing required command: {needed.command} - {needed.info}")
            missing_counter += 1
        else:
            log.log("✅", "info", f"{needed.command} available at {where}")

    return missing_counter

sparql_server

Created on 2025-05-27

@author: wf

Response

wrapper for responses including errors

Source code in omnigraph/sparql_server.py 
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
class Response:
    """
    wrapper for responses including errors
    """

    @property
    def success(self) -> bool:
        is_success = False
        if self.error is not None:
            is_success = False
        if self.response is not None:
            # HTTP status codes
            # 200 OK (request succeeded)
            # 201 Created (resource created)
            # 204 No Content (success with no response body)
            is_success = self.response.status_code in [200, 201, 204]
        return is_success

    def __init__(self, response=None, error=None):
        self.response = response
        self.error = error

SparqlServer

Base class for dockerized SPARQL servers

Source code in omnigraph/sparql_server.py 
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
class SparqlServer:
    """
    Base class for dockerized SPARQL servers
    """

    def __init__(self, config: ServerConfig, env: ServerEnv):
        """
        Initialize the SPARQL server manager.

        """
        self.env = env
        self.log = env.log
        self.config = config
        self.name = self.config.name
        self.debug = env.debug
        self.verbose = env.verbose
        self.shell = env.shell
        self.rdf_format = RdfFormat.by_label(self.config.rdf_format)
        self.current_status = None
        self.docker_util = DockerUtil(
            shell=self.shell,
            container_name=self.config.container_name,
            log=self.log,
            verbose=self.verbose,
            debug=self.debug,
        )

        # Subclasses must set these URLs
        if self.config.sparql_url:
            is_fuseki = self.config.server == "jena"
            self.sparql = SPARQL(self.config.sparql_url, isFuseki=is_fuseki)
            if (
                hasattr(self.config, "auth_password")
                and self.config.auth_password
                and hasattr(self.config, "auth_user")
                and self.config.auth_user
            ):
                self.sparql.addAuthentication(self.config.auth_user, self.config.auth_password)

    @property
    def full_name(self) -> str:
        full_name = f"{self.name} {self.config.container_name}"
        return full_name

    @property
    def flag(self) -> str:
        flag = "🟢️" if self.config.active else "🛑"
        if self.current_status:
            state = self.current_status.at.value
            flag += str(state)
        return flag

    def as_endpoint_conf(self, prefix_configs: PrefixConfigs, prefix_sets: List[str]) -> Endpoint:
        """
        Convert server configuration to Endpoint configuration.

        Args:
            prefix_configs: PrefixConfigs instance with prefix definitions
            prefix_sets: List of prefix set names to include

        Returns:
            Endpoint: Endpoint configuration object
        """
        endpoint = Endpoint()

        # Basic endpoint properties
        endpoint.name = self.config.name
        endpoint.lang = "sparql"
        endpoint.endpoint = self.config.sparql_url
        endpoint.website = self.config.web_url
        endpoint.database = self.config.server
        endpoint.method = "POST"

        # Authentication if configured
        if (
            hasattr(self.config, "auth_user")
            and self.config.auth_user
            and hasattr(self.config, "auth_password")
            and self.config.auth_password
        ):
            endpoint.auth = "BASIC"
            endpoint.user = self.config.auth_user
            endpoint.password = self.config.auth_password

        # Get prefixes from provided prefix sets
        declarations = prefix_configs.get_selected_declarations(prefix_sets)
        endpoint.prefixes = declarations

        return endpoint

    def avail_mem_gb(self) -> float:
        avail_mem = psutil.virtual_memory().available / (1024**3)
        return avail_mem

    def handle_exception(self, context: str, ex: Exception):
        """
        handle the given exception
        """
        container_name = self.config.container_name
        self.log.log("❌", container_name, f"[{self.full_name}] Exception {context}: {ex}")
        if self.debug:
            # extract exception type, and trace back
            ex_type = type(ex)
            ex_tb = ex.__traceback__
            # print exception stack details
            traceback.print_exception(ex_type, ex, ex_tb)

    def make_request(self, method: str, url: str, **kwargs) -> Response:
        """
        Helper function for making HTTP requests with consistent error handling.

        Args:
            method: HTTP method (GET, POST, etc.)
            url: Request URL
            **kwargs: Additional arguments for requests

        Returns:
            Response
        """
        try:
            #  add auth if we have auth_password and auth_user
            if (
                hasattr(self.config, "auth_password")
                and self.config.auth_password
                and hasattr(self.config, "auth_user")
                and self.config.auth_user
            ):
                kwargs.setdefault("auth", (self.config.auth_user, self.config.auth_password))
            # for Jena Fuseki we do this via url
            # Only set timeout if not already provided
            kwargs.setdefault("timeout", self.config.timeout)
            response = requests.request(method, url, **kwargs)
            response = Response(response)
        except Exception as ex:
            self.handle_exception(f"request {url}", ex)
            response = Response(None, ex)
        return response

    def get_web_url(self) -> str:
        """
        Return the service-specific Web UI URL.
        Subclasses may override.
        """
        return self.config.web_url

    def webui(self):
        """
        open my webui
        """
        web_url = self.get_web_url()
        webbrowser.open(web_url)

    def status_info(self) -> str:
        """
        Return one-line summary of server status e.g. for CLI use.
        """
        self.status()
        summary = self.current_status.get_summary(self.debug)
        info = f"{self.flag}{summary}"
        return info

    def status(self) -> ServerStatus:
        """
        Check server status using a single docker inspect call.

        Returns:
            ServerStatus: object with detailed container state
        """
        server_status = ServerStatus(at=ServerLifecycleState.UNKNOWN)
        state = self.docker_util.inspect()

        if state:
            server_status.exists = True
            server_status.running = state.get("Running", False)
            server_status.docker_status = state.get("Status")
            server_status.docker_exit_code = state.get("ExitCode")
            self.refresh_logs(server_status)
            if server_status.running:
                server_status.at = ServerLifecycleState.UP
                self.refresh_logs(server_status)
            else:
                if server_status.docker_status == "exited" and server_status.docker_exit_code not in (0, None):
                    server_status.at = ServerLifecycleState.ERROR
                elif server_status.docker_status == "created":
                    server_status.at = ServerLifecycleState.STARTING
                else:
                    server_status.at = ServerLifecycleState.STOPPED
        else:
            server_status.exists = False
            server_status.running = False

        self.current_status = server_status
        return server_status

    def refresh_logs(self, server_status=ServerStatus):
        """
        refresh the logs for the given server status
        """
        proc = self.shell.run(f"docker logs {self.config.container_name}", tee=False)
        logs = f"stdout:{proc.stdout}\nstderr:{proc.stderr}"
        server_status.logs = logs

    def add_triple_count2_server_status(self, server_status=ServerStatus):
        """
        add triple count to server status
        """
        try:
            triple_count = self.count_triples()
            server_status.triple_count = triple_count
        except Exception as ex:
            server_status.error = ex

    # delegates
    def run_shell_command(self, command: str, success_msg: str = None, error_msg: str = None) -> ShellResult:
        """
        Helper function for running shell commands with consistent error handling.
        """
        return self.docker_util.run_shell_command(command, success_msg, error_msg)

    def docker_cmd(self, cmd: str, options: str = "", args: str = "") -> str:
        """create the given docker command with the given options"""
        return self.docker_util.docker_cmd(cmd, options, args)

    def run_docker_cmd(self, cmd: str, options: str = "", args: str = "") -> ShellResult:
        """run the given docker commmand with the given options"""
        return self.docker_util.run_docker_cmd(cmd, options, args)

    def logs(self) -> ShellResult:
        """show the logs of the container"""
        logs = self.docker_util.logs()
        logs.success = not logs.proc.stderr
        # Print to respective streams
        print(logs.proc.stdout, file=sys.stdout, end="")
        print(logs.proc.stderr, file=sys.stderr, end="")

        return logs

    def docker_info(self) -> ShellResult:
        """Check if Docker is responsive on the host system."""
        return self.docker_util.docker_info()

    def stop(self) -> ShellResult:
        """stop the server container"""
        return self.docker_util.stop()

    def rm(self) -> ShellResult:
        """remove the server container."""
        return self.docker_util.rm()

    def bash(self) -> bool:
        """bash into the server container."""
        return self.docker_util.bash()

    def pre_create(self):
        """
        abstract pre docker create step
        implement a special version if need be
        """

    def post_create(self):
        """
        abstract post docker create step
        implement a special version if need be
        """

    def docker_create(self) -> bool:
        """
        Create and start a new Docker container for the configured server.

        Returns:
            bool: True if the container was created and is running, False otherwise.
        """
        container_name = self.config.container_name
        server_name = self.config.name
        self.log.log(
            "✅",
            container_name,
            f"Creating new {server_name} container {container_name}...",
        )
        try:
            self.pre_create()
            operation_success = True
        except Exception as ex:
            self.handle_exception("pre_create", ex)
            operation_success = False
        if operation_success:
            base_data_dir = self.config.base_data_dir
            create_cmd = self.config.get_docker_run_command(data_dir=base_data_dir)
            create_result = self.docker_util.run_shell_command(
                create_cmd,
                error_msg=f"Failed to create container {container_name}",
            )

            operation_success = create_result.success
            container_id = create_result.proc.stdout.strip()

            if not re.fullmatch(r"[0-9a-f]{12,}", container_id):
                self.log.log(
                    "❌",
                    container_name,
                    f"Creating new {server_name} container failed – invalid container ID '{container_id}' from command: {create_cmd}",
                )
                operation_success = False

        if operation_success:
            try:
                self.post_create()
            except Exception as ex:
                self.handle_exception("pre_create", ex)
                operation_success = False

        if operation_success:
            server_status = self.status()
            if not server_status.running:
                self.log.log(
                    "❌",
                    container_name,
                    f"Container exited with status='{server_status.docker_status}', exit_code={server_status.docker_exit_code}",
                )
                if server_status.logs:
                    self.log.log("ℹ️", container_name, f"Logs:\n{server_status.logs.strip()}")
                operation_success = False

        return operation_success

    def post_start(self, first_start: bool):
        """
        Abstract post start step.

        Args:
            first_start: True if this is the first time starting (new container)
        """
        pass

    def start(self, show_progress: bool = True) -> bool:
        """
        Start SPARQL server in Docker container.

        Args:
            show_progress: Show progress bar while waiting

        Returns:
            True if started successfully
        """
        container_name = self.config.container_name
        server_name = self.config.name
        # Check support status
        support = self.config.support_status
        support.log_status(self.log, container_name, self.config)

        start_success = False
        first_start = False  # Track if this is first start

        if not support.is_blocking():
            try:
                docker_status = self.docker_info()
                operation_success = docker_status.success
                if operation_success:
                    server_status = self.status()

                    if server_status.running:
                        self.log.log(
                            "✅",
                            container_name,
                            f"Container {container_name} is already running",
                        )
                        operation_success = True
                    elif server_status.exists:
                        self.log.log(
                            "✅",
                            container_name,
                            f"Container {container_name} exists, starting...",
                        )
                        start_cmd = f"docker start {container_name}"
                        start_result = self.docker_util.run_shell_command(
                            start_cmd,
                            error_msg=f"Failed to start container {container_name}",
                        )
                        operation_success = start_result
                    else:
                        operation_success = self.docker_create()
                        first_start = True

                if operation_success:
                    start_success = self.wait_until_ready(show_progress=show_progress)
                    if start_success:
                        self.post_start(first_start)
                else:
                    start_success = False

            except Exception as ex:
                self.handle_exception(f"starting {server_name}", ex)
                start_success = False
        return start_success

    def count_triples(self) -> int:
        """
        Count total triples in the SPARQL server.

        Returns:
            Number of triples
        """
        count_query = "SELECT (COUNT(*) AS ?count) WHERE { ?s ?p ?o }"
        try:
            result = self.sparql.getValue(count_query, "count")
            triple_count = int(result) if result else 0
        except Exception as ex:
            self.handle_exception("count_triples", ex)
            triple_count = -1
        return triple_count

    def wait_until_ready(self, show_progress: bool = False) -> bool:
        """
        Wait for server to be ready.

        Args:
            timeout: Maximum seconds to wait
            show_progress: Show progress bar while waiting

        Returns:
            True if ready within timeout
        """
        container_name = self.config.container_name
        base_url = self.config.base_url
        timeout = self.config.ready_timeout

        self.log.log(
            "✅",
            container_name,
            f"Waiting for {self.full_name} to start ... ",
        )

        pbar = None
        if show_progress:
            pbar = tqdm(total=timeout, desc=f"Waiting for {self.full_name}", unit="s")

        ready_status = False
        for secs in range(timeout):
            server_status = self.status()
            if server_status.at == ServerLifecycleState.READY:
                if show_progress and pbar:
                    pbar.close()
                self.log.log(
                    "✅",
                    container_name,
                    f"{self.full_name} ready at {base_url} after {secs}s",
                )
                ready_status = True
                break

            if show_progress and pbar:
                pbar.update(1)
            time.sleep(1)

        if not ready_status:
            if show_progress and pbar:
                pbar.close()
            self.log.log(
                "⚠️",
                container_name,
                f"Timeout waiting for {self.full_name} to start after {timeout}s",
            )

        return ready_status

    def get_clear_query(self) -> str:
        """
        the clear query to be used
        may be overriden by specific SPARQL server implementations
        """
        # clear_query = "DELETE { ?s ?p ?o } WHERE { ?s ?p ?o }"
        clear_query = "CLEAR ALL"
        return clear_query

    def execute_update_query_with_post(self, update_query: str) -> tuple[Optional[Any], Optional[Exception]]:
        """
        Execute SPARQL UPDATE query
        using application/sparql-update content type for UPDATE operations.

        Args:
            update_query: SPARQL UPDATE query string

        Returns:
            Tuple of (response, exception)
        """
        result: Optional[Any] = None
        error: Optional[Exception] = None

        try:
            resp = self.make_request(
                "POST",
                self.config.update_url,
                headers={"Content-Type": "application/sparql-update"},
                data=update_query,
                timeout=self.config.upload_timeout,
            )

            result = resp.response
            if not resp.success:
                status = resp.response.status_code if resp.response else "unknown"
                error = resp.error or Exception(f"HTTP {status}")

        except Exception as ex:
            error = ex

        return result, error

    def execute_update_query(self, update_query: str) -> tuple[any, Exception]:
        """
        Execute a SPARQL UPDATE query (INSERT/DELETE).

        This method can be overridden by subclasses to handle server-specific
        requirements (e.g., different content types, endpoints).

        Args:
            update_query: SPARQL UPDATE query string

        Returns:
            Tuple of (response, exception)
        """
        return self.sparql.insert(update_query)

    def clear(self) -> int:
        """
        Delete all triples.
        """
        container_name = self.config.container_name
        count_triples = self.count_triples()
        msg = f"deleting {count_triples} triples ..."
        protected = count_triples >= self.config.unforced_clear_limit

        if protected and not self.env.force:
            self.log.log("❌", container_name, f"{msg} needs force option")
        else:
            clear_query = self.get_clear_query()
            try:
                _response, ex = self.execute_update_query(clear_query)
                if ex:
                    self.handle_exception("DELETE", ex)
                new_count = self.count_triples()
                if new_count == 0:
                    self.log.log("✅", container_name, f"deleted {count_triples} triples")
                else:
                    self.log.log(
                        "❌",
                        container_name,
                        f"[{self.full_name}] delete failed: {new_count} triples remain",
                    )
                count_triples = new_count
            except Exception as ex:
                self.handle_exception("clear triples", ex)

        return count_triples

    def upload_request(self, file_content: bytes) -> Response:
        """Default upload request for Blazegraph-style servers."""
        response = self.make_request(
            "POST",
            self.config.upload_url,
            headers={"Content-Type": self.rdf_format.mime_type},
            data=file_content,
            timeout=self.config.upload_timeout,
        )
        return response

    def load_file(self, filepath: str, upload_request=None) -> bool:
        """
        Load a single RDF file into the RDF server.
        """
        container_name = self.config.container_name
        load_success = False

        if upload_request is None:
            upload_request_callback = self.upload_request
        else:
            upload_request_callback = upload_request

        try:
            with open(filepath, "rb") as f:
                file_content = f.read()

            response = upload_request_callback(file_content)

            if response.success:  # Changed from result["success"]
                self.log.log("✅", container_name, f"Loaded {filepath}")
                load_success = True
            else:
                if response.error:
                    error_msg = str(response.error)
                else:
                    status_code = response.response.status_code
                    content = response.response.text
                    error_msg = f"HTTP {status_code} → {content}"
                self.log.log("❌", container_name, f"Failed to load {filepath}: {error_msg}")
                load_success = False

        except Exception as ex:
            self.handle_exception(f"loading {filepath}", ex)
            load_success = False

        return load_success

    def get_dump_files(self, file_pattern: str = None) -> List[Path]:
        """
        Get the dump files matching the given pattern.

        Args:
            file_pattern: Glob pattern for dump files

        Returns:
            sorted list of matching dump file paths
        """
        dump_path: Path = Path(self.config.dumps_dir)
        if file_pattern is None:
            file_pattern = f"*{self.rdf_format.extension}"
        files = sorted(dump_path.glob(file_pattern))
        return files

    def upload_dump_files(self, file_pattern: str = None) -> int:
        """
        Bulk-upload all dump files matching pattern using the server's
        native bulk load mechanism.

        Subclasses override this with their native loader (e.g. Jena
        tdb2.tdbloader, Blazegraph REST DataLoader, QLever index build);
        the default falls back to the HTTP load path.

        Args:
            file_pattern: Glob pattern for dump files

        Returns:
            Number of files loaded successfully
        """
        loaded_count = self.load_dump_files(file_pattern)
        return loaded_count

    def load_dump_files(self, file_pattern: str = None) -> int:
        """
        Load all dump files matching pattern.

        Args:
            file_pattern: Glob pattern for dump files
            use_bulk: Use bulk loader if True, individual files if False

        Returns:
            Number of files loaded successfully
        """
        files = self.get_dump_files(file_pattern)
        loaded_count = 0
        container_name = self.config.container_name

        if not files:
            self.log.log("⚠️", container_name, f"No files found matching pattern: {file_pattern}")
        else:
            self.log.log("✅", container_name, f"Found {len(files)} files to load")
            pbar = tqdm(files, dynamic_ncols=True)
            for filepath in pbar:
                pbar.set_description(f"Mem: {self.avail_mem_gb():.1f} GB → {filepath.name}")
                file_result = self.load_file(filepath)
                if file_result:
                    loaded_count += 1
                else:
                    self.log.log("❌", container_name, f"Failed to load: {filepath}")

        return loaded_count

    def check_needed_software(self) -> int:
        """
        Check if needed software for this server configuration is installed
        """
        container_name = self.config.container_name
        if self.config.needed_software is None:
            return
        software_list = SoftwareList.from_dict2(self.config.needed_software)  # @UndefinedVariable
        missing = software_list.check_installed(self.log, self.shell, verbose=True)
        if missing > 0:
            self.log.log(
                "❌",
                container_name,
                "Please install the missing commands before running this script.",
            )
        return missing

__init__(config, env)

Initialize the SPARQL server manager.

Source code in omnigraph/sparql_server.py 
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
def __init__(self, config: ServerConfig, env: ServerEnv):
    """
    Initialize the SPARQL server manager.

    """
    self.env = env
    self.log = env.log
    self.config = config
    self.name = self.config.name
    self.debug = env.debug
    self.verbose = env.verbose
    self.shell = env.shell
    self.rdf_format = RdfFormat.by_label(self.config.rdf_format)
    self.current_status = None
    self.docker_util = DockerUtil(
        shell=self.shell,
        container_name=self.config.container_name,
        log=self.log,
        verbose=self.verbose,
        debug=self.debug,
    )

    # Subclasses must set these URLs
    if self.config.sparql_url:
        is_fuseki = self.config.server == "jena"
        self.sparql = SPARQL(self.config.sparql_url, isFuseki=is_fuseki)
        if (
            hasattr(self.config, "auth_password")
            and self.config.auth_password
            and hasattr(self.config, "auth_user")
            and self.config.auth_user
        ):
            self.sparql.addAuthentication(self.config.auth_user, self.config.auth_password)

add_triple_count2_server_status(server_status=ServerStatus)

add triple count to server status

Source code in omnigraph/sparql_server.py 
290
291
292
293
294
295
296
297
298
def add_triple_count2_server_status(self, server_status=ServerStatus):
    """
    add triple count to server status
    """
    try:
        triple_count = self.count_triples()
        server_status.triple_count = triple_count
    except Exception as ex:
        server_status.error = ex

as_endpoint_conf(prefix_configs, prefix_sets)

Convert server configuration to Endpoint configuration.

Parameters:

Name Type Description Default
prefix_configs PrefixConfigs

PrefixConfigs instance with prefix definitions

 required
prefix_sets List[str]

List of prefix set names to include

 required

Returns:

Name Type Description
Endpoint Endpoint

Endpoint configuration object
Source code in omnigraph/sparql_server.py 
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
def as_endpoint_conf(self, prefix_configs: PrefixConfigs, prefix_sets: List[str]) -> Endpoint:
    """
    Convert server configuration to Endpoint configuration.

    Args:
        prefix_configs: PrefixConfigs instance with prefix definitions
        prefix_sets: List of prefix set names to include

    Returns:
        Endpoint: Endpoint configuration object
    """
    endpoint = Endpoint()

    # Basic endpoint properties
    endpoint.name = self.config.name
    endpoint.lang = "sparql"
    endpoint.endpoint = self.config.sparql_url
    endpoint.website = self.config.web_url
    endpoint.database = self.config.server
    endpoint.method = "POST"

    # Authentication if configured
    if (
        hasattr(self.config, "auth_user")
        and self.config.auth_user
        and hasattr(self.config, "auth_password")
        and self.config.auth_password
    ):
        endpoint.auth = "BASIC"
        endpoint.user = self.config.auth_user
        endpoint.password = self.config.auth_password

    # Get prefixes from provided prefix sets
    declarations = prefix_configs.get_selected_declarations(prefix_sets)
    endpoint.prefixes = declarations

    return endpoint

bash()

bash into the server container.

Source code in omnigraph/sparql_server.py 
337
338
339
def bash(self) -> bool:
    """bash into the server container."""
    return self.docker_util.bash()

check_needed_software()

Check if needed software for this server configuration is installed

Source code in omnigraph/sparql_server.py 
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
def check_needed_software(self) -> int:
    """
    Check if needed software for this server configuration is installed
    """
    container_name = self.config.container_name
    if self.config.needed_software is None:
        return
    software_list = SoftwareList.from_dict2(self.config.needed_software)  # @UndefinedVariable
    missing = software_list.check_installed(self.log, self.shell, verbose=True)
    if missing > 0:
        self.log.log(
            "❌",
            container_name,
            "Please install the missing commands before running this script.",
        )
    return missing

clear()

Delete all triples.

Source code in omnigraph/sparql_server.py 
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
def clear(self) -> int:
    """
    Delete all triples.
    """
    container_name = self.config.container_name
    count_triples = self.count_triples()
    msg = f"deleting {count_triples} triples ..."
    protected = count_triples >= self.config.unforced_clear_limit

    if protected and not self.env.force:
        self.log.log("❌", container_name, f"{msg} needs force option")
    else:
        clear_query = self.get_clear_query()
        try:
            _response, ex = self.execute_update_query(clear_query)
            if ex:
                self.handle_exception("DELETE", ex)
            new_count = self.count_triples()
            if new_count == 0:
                self.log.log("✅", container_name, f"deleted {count_triples} triples")
            else:
                self.log.log(
                    "❌",
                    container_name,
                    f"[{self.full_name}] delete failed: {new_count} triples remain",
                )
            count_triples = new_count
        except Exception as ex:
            self.handle_exception("clear triples", ex)

    return count_triples

count_triples()

Count total triples in the SPARQL server.

Returns:

Type Description
int

Number of triples
Source code in omnigraph/sparql_server.py 
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
def count_triples(self) -> int:
    """
    Count total triples in the SPARQL server.

    Returns:
        Number of triples
    """
    count_query = "SELECT (COUNT(*) AS ?count) WHERE { ?s ?p ?o }"
    try:
        result = self.sparql.getValue(count_query, "count")
        triple_count = int(result) if result else 0
    except Exception as ex:
        self.handle_exception("count_triples", ex)
        triple_count = -1
    return triple_count

docker_cmd(cmd, options='', args='')

create the given docker command with the given options

Source code in omnigraph/sparql_server.py 
307
308
309
def docker_cmd(self, cmd: str, options: str = "", args: str = "") -> str:
    """create the given docker command with the given options"""
    return self.docker_util.docker_cmd(cmd, options, args)

docker_create()

Create and start a new Docker container for the configured server.

Returns:

Name Type Description
bool bool

True if the container was created and is running, False otherwise.
Source code in omnigraph/sparql_server.py 
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
def docker_create(self) -> bool:
    """
    Create and start a new Docker container for the configured server.

    Returns:
        bool: True if the container was created and is running, False otherwise.
    """
    container_name = self.config.container_name
    server_name = self.config.name
    self.log.log(
        "✅",
        container_name,
        f"Creating new {server_name} container {container_name}...",
    )
    try:
        self.pre_create()
        operation_success = True
    except Exception as ex:
        self.handle_exception("pre_create", ex)
        operation_success = False
    if operation_success:
        base_data_dir = self.config.base_data_dir
        create_cmd = self.config.get_docker_run_command(data_dir=base_data_dir)
        create_result = self.docker_util.run_shell_command(
            create_cmd,
            error_msg=f"Failed to create container {container_name}",
        )

        operation_success = create_result.success
        container_id = create_result.proc.stdout.strip()

        if not re.fullmatch(r"[0-9a-f]{12,}", container_id):
            self.log.log(
                "❌",
                container_name,
                f"Creating new {server_name} container failed – invalid container ID '{container_id}' from command: {create_cmd}",
            )
            operation_success = False

    if operation_success:
        try:
            self.post_create()
        except Exception as ex:
            self.handle_exception("pre_create", ex)
            operation_success = False

    if operation_success:
        server_status = self.status()
        if not server_status.running:
            self.log.log(
                "❌",
                container_name,
                f"Container exited with status='{server_status.docker_status}', exit_code={server_status.docker_exit_code}",
            )
            if server_status.logs:
                self.log.log("ℹ️", container_name, f"Logs:\n{server_status.logs.strip()}")
            operation_success = False

    return operation_success

docker_info()

Check if Docker is responsive on the host system.

Source code in omnigraph/sparql_server.py 
325
326
327
def docker_info(self) -> ShellResult:
    """Check if Docker is responsive on the host system."""
    return self.docker_util.docker_info()

execute_update_query(update_query)

Execute a SPARQL UPDATE query (INSERT/DELETE).

This method can be overridden by subclasses to handle server-specific requirements (e.g., different content types, endpoints).

Parameters:

Name Type Description Default
update_query str

SPARQL UPDATE query string

 required

Returns:

Type Description
tuple[any, Exception]

Tuple of (response, exception)
Source code in omnigraph/sparql_server.py 
595
596
597
598
599
600
601
602
603
604
605
606
607
608
def execute_update_query(self, update_query: str) -> tuple[any, Exception]:
    """
    Execute a SPARQL UPDATE query (INSERT/DELETE).

    This method can be overridden by subclasses to handle server-specific
    requirements (e.g., different content types, endpoints).

    Args:
        update_query: SPARQL UPDATE query string

    Returns:
        Tuple of (response, exception)
    """
    return self.sparql.insert(update_query)

execute_update_query_with_post(update_query)

Execute SPARQL UPDATE query using application/sparql-update content type for UPDATE operations.

Parameters:

Name Type Description Default
update_query str

SPARQL UPDATE query string

 required

Returns:

Type Description
tuple[Optional[Any], Optional[Exception]]

Tuple of (response, exception)
Source code in omnigraph/sparql_server.py 
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
def execute_update_query_with_post(self, update_query: str) -> tuple[Optional[Any], Optional[Exception]]:
    """
    Execute SPARQL UPDATE query
    using application/sparql-update content type for UPDATE operations.

    Args:
        update_query: SPARQL UPDATE query string

    Returns:
        Tuple of (response, exception)
    """
    result: Optional[Any] = None
    error: Optional[Exception] = None

    try:
        resp = self.make_request(
            "POST",
            self.config.update_url,
            headers={"Content-Type": "application/sparql-update"},
            data=update_query,
            timeout=self.config.upload_timeout,
        )

        result = resp.response
        if not resp.success:
            status = resp.response.status_code if resp.response else "unknown"
            error = resp.error or Exception(f"HTTP {status}")

    except Exception as ex:
        error = ex

    return result, error

get_clear_query()

the clear query to be used may be overriden by specific SPARQL server implementations

Source code in omnigraph/sparql_server.py 
553
554
555
556
557
558
559
560
def get_clear_query(self) -> str:
    """
    the clear query to be used
    may be overriden by specific SPARQL server implementations
    """
    # clear_query = "DELETE { ?s ?p ?o } WHERE { ?s ?p ?o }"
    clear_query = "CLEAR ALL"
    return clear_query

get_dump_files(file_pattern=None)

Get the dump files matching the given pattern.

Parameters:

Name Type Description Default
file_pattern str

Glob pattern for dump files

 None

Returns:

Type Description
List[Path]

sorted list of matching dump file paths
Source code in omnigraph/sparql_server.py 
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
def get_dump_files(self, file_pattern: str = None) -> List[Path]:
    """
    Get the dump files matching the given pattern.

    Args:
        file_pattern: Glob pattern for dump files

    Returns:
        sorted list of matching dump file paths
    """
    dump_path: Path = Path(self.config.dumps_dir)
    if file_pattern is None:
        file_pattern = f"*{self.rdf_format.extension}"
    files = sorted(dump_path.glob(file_pattern))
    return files

get_web_url()

Return the service-specific Web UI URL. Subclasses may override.

Source code in omnigraph/sparql_server.py 
226
227
228
229
230
231
def get_web_url(self) -> str:
    """
    Return the service-specific Web UI URL.
    Subclasses may override.
    """
    return self.config.web_url

handle_exception(context, ex)

handle the given exception

Source code in omnigraph/sparql_server.py 
182
183
184
185
186
187
188
189
190
191
192
193
def handle_exception(self, context: str, ex: Exception):
    """
    handle the given exception
    """
    container_name = self.config.container_name
    self.log.log("❌", container_name, f"[{self.full_name}] Exception {context}: {ex}")
    if self.debug:
        # extract exception type, and trace back
        ex_type = type(ex)
        ex_tb = ex.__traceback__
        # print exception stack details
        traceback.print_exception(ex_type, ex, ex_tb)

load_dump_files(file_pattern=None)

Load all dump files matching pattern.

Parameters:

Name Type Description Default
file_pattern str

Glob pattern for dump files

 None
use_bulk

Use bulk loader if True, individual files if False

 required

Returns:

Type Description
int

Number of files loaded successfully
Source code in omnigraph/sparql_server.py 
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
def load_dump_files(self, file_pattern: str = None) -> int:
    """
    Load all dump files matching pattern.

    Args:
        file_pattern: Glob pattern for dump files
        use_bulk: Use bulk loader if True, individual files if False

    Returns:
        Number of files loaded successfully
    """
    files = self.get_dump_files(file_pattern)
    loaded_count = 0
    container_name = self.config.container_name

    if not files:
        self.log.log("⚠️", container_name, f"No files found matching pattern: {file_pattern}")
    else:
        self.log.log("✅", container_name, f"Found {len(files)} files to load")
        pbar = tqdm(files, dynamic_ncols=True)
        for filepath in pbar:
            pbar.set_description(f"Mem: {self.avail_mem_gb():.1f} GB → {filepath.name}")
            file_result = self.load_file(filepath)
            if file_result:
                loaded_count += 1
            else:
                self.log.log("❌", container_name, f"Failed to load: {filepath}")

    return loaded_count

load_file(filepath, upload_request=None)

Load a single RDF file into the RDF server.

Source code in omnigraph/sparql_server.py 
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
def load_file(self, filepath: str, upload_request=None) -> bool:
    """
    Load a single RDF file into the RDF server.
    """
    container_name = self.config.container_name
    load_success = False

    if upload_request is None:
        upload_request_callback = self.upload_request
    else:
        upload_request_callback = upload_request

    try:
        with open(filepath, "rb") as f:
            file_content = f.read()

        response = upload_request_callback(file_content)

        if response.success:  # Changed from result["success"]
            self.log.log("✅", container_name, f"Loaded {filepath}")
            load_success = True
        else:
            if response.error:
                error_msg = str(response.error)
            else:
                status_code = response.response.status_code
                content = response.response.text
                error_msg = f"HTTP {status_code} → {content}"
            self.log.log("❌", container_name, f"Failed to load {filepath}: {error_msg}")
            load_success = False

    except Exception as ex:
        self.handle_exception(f"loading {filepath}", ex)
        load_success = False

    return load_success

logs()

show the logs of the container

Source code in omnigraph/sparql_server.py 
315
316
317
318
319
320
321
322
323
def logs(self) -> ShellResult:
    """show the logs of the container"""
    logs = self.docker_util.logs()
    logs.success = not logs.proc.stderr
    # Print to respective streams
    print(logs.proc.stdout, file=sys.stdout, end="")
    print(logs.proc.stderr, file=sys.stderr, end="")

    return logs

make_request(method, url, **kwargs)

Helper function for making HTTP requests with consistent error handling.

Parameters:

Name Type Description Default
method str

HTTP method (GET, POST, etc.)

 required
url str

Request URL

 required
**kwargs

Additional arguments for requests

 {}

Returns:

Type Description
Response

Response
Source code in omnigraph/sparql_server.py 
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
def make_request(self, method: str, url: str, **kwargs) -> Response:
    """
    Helper function for making HTTP requests with consistent error handling.

    Args:
        method: HTTP method (GET, POST, etc.)
        url: Request URL
        **kwargs: Additional arguments for requests

    Returns:
        Response
    """
    try:
        #  add auth if we have auth_password and auth_user
        if (
            hasattr(self.config, "auth_password")
            and self.config.auth_password
            and hasattr(self.config, "auth_user")
            and self.config.auth_user
        ):
            kwargs.setdefault("auth", (self.config.auth_user, self.config.auth_password))
        # for Jena Fuseki we do this via url
        # Only set timeout if not already provided
        kwargs.setdefault("timeout", self.config.timeout)
        response = requests.request(method, url, **kwargs)
        response = Response(response)
    except Exception as ex:
        self.handle_exception(f"request {url}", ex)
        response = Response(None, ex)
    return response

post_create()

abstract post docker create step implement a special version if need be

Source code in omnigraph/sparql_server.py 
347
348
349
350
351
def post_create(self):
    """
    abstract post docker create step
    implement a special version if need be
    """

post_start(first_start)

Abstract post start step.

Parameters:

Name Type Description Default
first_start bool

True if this is the first time starting (new container)

 required
Source code in omnigraph/sparql_server.py 
413
414
415
416
417
418
419
420
def post_start(self, first_start: bool):
    """
    Abstract post start step.

    Args:
        first_start: True if this is the first time starting (new container)
    """
    pass

pre_create()

abstract pre docker create step implement a special version if need be

Source code in omnigraph/sparql_server.py 
341
342
343
344
345
def pre_create(self):
    """
    abstract pre docker create step
    implement a special version if need be
    """

refresh_logs(server_status=ServerStatus)

refresh the logs for the given server status

Source code in omnigraph/sparql_server.py 
282
283
284
285
286
287
288
def refresh_logs(self, server_status=ServerStatus):
    """
    refresh the logs for the given server status
    """
    proc = self.shell.run(f"docker logs {self.config.container_name}", tee=False)
    logs = f"stdout:{proc.stdout}\nstderr:{proc.stderr}"
    server_status.logs = logs

rm()

remove the server container.

Source code in omnigraph/sparql_server.py 
333
334
335
def rm(self) -> ShellResult:
    """remove the server container."""
    return self.docker_util.rm()

run_docker_cmd(cmd, options='', args='')

run the given docker commmand with the given options

Source code in omnigraph/sparql_server.py 
311
312
313
def run_docker_cmd(self, cmd: str, options: str = "", args: str = "") -> ShellResult:
    """run the given docker commmand with the given options"""
    return self.docker_util.run_docker_cmd(cmd, options, args)

run_shell_command(command, success_msg=None, error_msg=None)

Helper function for running shell commands with consistent error handling.

Source code in omnigraph/sparql_server.py 
301
302
303
304
305
def run_shell_command(self, command: str, success_msg: str = None, error_msg: str = None) -> ShellResult:
    """
    Helper function for running shell commands with consistent error handling.
    """
    return self.docker_util.run_shell_command(command, success_msg, error_msg)

start(show_progress=True)

Start SPARQL server in Docker container.

Parameters:

Name Type Description Default
show_progress bool

Show progress bar while waiting

 True

Returns:

Type Description
bool

True if started successfully
Source code in omnigraph/sparql_server.py 
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
def start(self, show_progress: bool = True) -> bool:
    """
    Start SPARQL server in Docker container.

    Args:
        show_progress: Show progress bar while waiting

    Returns:
        True if started successfully
    """
    container_name = self.config.container_name
    server_name = self.config.name
    # Check support status
    support = self.config.support_status
    support.log_status(self.log, container_name, self.config)

    start_success = False
    first_start = False  # Track if this is first start

    if not support.is_blocking():
        try:
            docker_status = self.docker_info()
            operation_success = docker_status.success
            if operation_success:
                server_status = self.status()

                if server_status.running:
                    self.log.log(
                        "✅",
                        container_name,
                        f"Container {container_name} is already running",
                    )
                    operation_success = True
                elif server_status.exists:
                    self.log.log(
                        "✅",
                        container_name,
                        f"Container {container_name} exists, starting...",
                    )
                    start_cmd = f"docker start {container_name}"
                    start_result = self.docker_util.run_shell_command(
                        start_cmd,
                        error_msg=f"Failed to start container {container_name}",
                    )
                    operation_success = start_result
                else:
                    operation_success = self.docker_create()
                    first_start = True

            if operation_success:
                start_success = self.wait_until_ready(show_progress=show_progress)
                if start_success:
                    self.post_start(first_start)
            else:
                start_success = False

        except Exception as ex:
            self.handle_exception(f"starting {server_name}", ex)
            start_success = False
    return start_success

status()

Check server status using a single docker inspect call.

Returns:

Name Type Description
ServerStatus ServerStatus

object with detailed container state
Source code in omnigraph/sparql_server.py 
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
def status(self) -> ServerStatus:
    """
    Check server status using a single docker inspect call.

    Returns:
        ServerStatus: object with detailed container state
    """
    server_status = ServerStatus(at=ServerLifecycleState.UNKNOWN)
    state = self.docker_util.inspect()

    if state:
        server_status.exists = True
        server_status.running = state.get("Running", False)
        server_status.docker_status = state.get("Status")
        server_status.docker_exit_code = state.get("ExitCode")
        self.refresh_logs(server_status)
        if server_status.running:
            server_status.at = ServerLifecycleState.UP
            self.refresh_logs(server_status)
        else:
            if server_status.docker_status == "exited" and server_status.docker_exit_code not in (0, None):
                server_status.at = ServerLifecycleState.ERROR
            elif server_status.docker_status == "created":
                server_status.at = ServerLifecycleState.STARTING
            else:
                server_status.at = ServerLifecycleState.STOPPED
    else:
        server_status.exists = False
        server_status.running = False

    self.current_status = server_status
    return server_status

status_info()

Return one-line summary of server status e.g. for CLI use.

Source code in omnigraph/sparql_server.py 
240
241
242
243
244
245
246
247
def status_info(self) -> str:
    """
    Return one-line summary of server status e.g. for CLI use.
    """
    self.status()
    summary = self.current_status.get_summary(self.debug)
    info = f"{self.flag}{summary}"
    return info

stop()

stop the server container

Source code in omnigraph/sparql_server.py 
329
330
331
def stop(self) -> ShellResult:
    """stop the server container"""
    return self.docker_util.stop()

upload_dump_files(file_pattern=None)

Bulk-upload all dump files matching pattern using the server's native bulk load mechanism.

Subclasses override this with their native loader (e.g. Jena tdb2.tdbloader, Blazegraph REST DataLoader, QLever index build); the default falls back to the HTTP load path.

Parameters:

Name Type Description Default
file_pattern str

Glob pattern for dump files

 None

Returns:

Type Description
int

Number of files loaded successfully
Source code in omnigraph/sparql_server.py 
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
def upload_dump_files(self, file_pattern: str = None) -> int:
    """
    Bulk-upload all dump files matching pattern using the server's
    native bulk load mechanism.

    Subclasses override this with their native loader (e.g. Jena
    tdb2.tdbloader, Blazegraph REST DataLoader, QLever index build);
    the default falls back to the HTTP load path.

    Args:
        file_pattern: Glob pattern for dump files

    Returns:
        Number of files loaded successfully
    """
    loaded_count = self.load_dump_files(file_pattern)
    return loaded_count

upload_request(file_content)

Default upload request for Blazegraph-style servers.

Source code in omnigraph/sparql_server.py 
642
643
644
645
646
647
648
649
650
651
def upload_request(self, file_content: bytes) -> Response:
    """Default upload request for Blazegraph-style servers."""
    response = self.make_request(
        "POST",
        self.config.upload_url,
        headers={"Content-Type": self.rdf_format.mime_type},
        data=file_content,
        timeout=self.config.upload_timeout,
    )
    return response

wait_until_ready(show_progress=False)

Wait for server to be ready.

Parameters:

Name Type Description Default
timeout

Maximum seconds to wait

 required
show_progress bool

Show progress bar while waiting

 False

Returns:

Type Description
bool

True if ready within timeout
Source code in omnigraph/sparql_server.py 
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
def wait_until_ready(self, show_progress: bool = False) -> bool:
    """
    Wait for server to be ready.

    Args:
        timeout: Maximum seconds to wait
        show_progress: Show progress bar while waiting

    Returns:
        True if ready within timeout
    """
    container_name = self.config.container_name
    base_url = self.config.base_url
    timeout = self.config.ready_timeout

    self.log.log(
        "✅",
        container_name,
        f"Waiting for {self.full_name} to start ... ",
    )

    pbar = None
    if show_progress:
        pbar = tqdm(total=timeout, desc=f"Waiting for {self.full_name}", unit="s")

    ready_status = False
    for secs in range(timeout):
        server_status = self.status()
        if server_status.at == ServerLifecycleState.READY:
            if show_progress and pbar:
                pbar.close()
            self.log.log(
                "✅",
                container_name,
                f"{self.full_name} ready at {base_url} after {secs}s",
            )
            ready_status = True
            break

        if show_progress and pbar:
            pbar.update(1)
        time.sleep(1)

    if not ready_status:
        if show_progress and pbar:
            pbar.close()
        self.log.log(
            "⚠️",
            container_name,
            f"Timeout waiting for {self.full_name} to start after {timeout}s",
        )

    return ready_status

webui()

open my webui

Source code in omnigraph/sparql_server.py 
233
234
235
236
237
238
def webui(self):
    """
    open my webui
    """
    web_url = self.get_web_url()
    webbrowser.open(web_url)

Step dataclass

a setup step

Source code in omnigraph/sparql_server.py 
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
@dataclass
class Step:
    """
    a setup step
    """

    name: str
    data_dir: Path
    setup_cmd: Optional[str] = None
    file_name: Optional[str] = None
    step: int = 0
    success: bool = False

    @property
    def path(self) -> Optional[Path]:
        if self.file_name:
            return self.data_dir / self.file_name
        return None

    def perform(self, server: "SparqlServer"):
        """
        perform the setup_cmd if self.path is not created yet
        """
        if self.path and self.path.exists():
            self.success = True
            msg = f"{self.path} already exists"
            server.log.log("✅", self.name, msg)
        else:
            command = f"cd {self.data_dir};{self.setup_cmd}"
            success_msg = f"{self.name} done"
            error_msg = f"{self.name} failed"
            shell_result = server.run_shell_command(command, success_msg, error_msg)
            self.success = shell_result.success

perform(server)

perform the setup_cmd if self.path is not created yet

Source code in omnigraph/sparql_server.py 
72
73
74
75
76
77
78
79
80
81
82
83
84
85
def perform(self, server: "SparqlServer"):
    """
    perform the setup_cmd if self.path is not created yet
    """
    if self.path and self.path.exists():
        self.success = True
        msg = f"{self.path} already exists"
        server.log.log("✅", self.name, msg)
    else:
        command = f"cd {self.data_dir};{self.setup_cmd}"
        success_msg = f"{self.name} done"
        error_msg = f"{self.name} failed"
        shell_result = server.run_shell_command(command, success_msg, error_msg)
        self.success = shell_result.success

version

Created on 2025-05-28

@author: wf

Version

Version handling for nicegui widgets

Source code in omnigraph/version.py 
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
@lod_storable
class Version:
    """
    Version handling for nicegui widgets
    """

    name = "omnigraph"
    version = omnigraph.__version__
    date = "2025-11-01"
    updated = "2026-03-15"
    description = "Unified Python interface for multiple graph databases"

    authors = "Wolfgang Fahl"

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

    license = f"""Copyright 2025 contributors. All rights reserved.

  Licensed under the Apache License 2.0
  http://www.apache.org/licenses/LICENSE-2.0

  Distributed on an "AS IS" basis without warranties
  or conditions of any kind, either express or implied."""

    longDescription = f"""{name} version {version}
{description}

  Created by {authors} on {date} last updated {updated}"""