misc.python.materialize.optbench 

 1# Copyright Materialize, Inc. and contributors. All rights reserved.
 2#
 3# Use of this software is governed by the Business Source License
 4# included in the LICENSE file at the root of this repository.
 5#
 6# As of the Change Date specified in that file, in accordance with
 7# the Business Source License, use of this software will be governed
 8# by the Apache License, Version 2.0.
 9
10from importlib import resources
11from pathlib import Path
12from typing import cast
13
14
15def resource_path(name: str) -> Path:
16    # NOTE: we have to do this cast because pyright is not comfortable with the
17    # Traversable protocol.
18    return cast(Path, resources.files(__package__)) / name
19
20
21def scenarios() -> list[str]:
22    """
23    Determines a list of avilable scenarios based on the intersection
24    of files located in both the `schema` and `workload` resource paths.
25    """
26    schema_files = {
27        p.name.removesuffix(".sql")
28        for p in resource_path("schema").iterdir()
29        if p.is_file() and p.name.endswith(".sql")
30    }
31    workload_files = {
32        p.name.removesuffix(".sql")
33        for p in resource_path("workload").iterdir()
34        if p.is_file() and p.name.endswith(".sql")
35    }
36
37    return sorted(schema_files.intersection(workload_files))
38
39
40class Scenario:
41    def __init__(self, value: str) -> None:
42        self.value = value
43
44    def schema_path(self) -> Path:
45        return resource_path(f"schema/{self}.sql")
46
47    def workload_path(self) -> Path:
48        return resource_path(f"workload/{self}.sql")
49
50    def __str__(self) -> str:
51        return self.value
def resource_path(name: str) -> pathlib.Path: 
16def resource_path(name: str) -> Path:
17    # NOTE: we have to do this cast because pyright is not comfortable with the
18    # Traversable protocol.
19    return cast(Path, resources.files(__package__)) / name
def scenarios() -> list[str]: 
22def scenarios() -> list[str]:
23    """
24    Determines a list of avilable scenarios based on the intersection
25    of files located in both the `schema` and `workload` resource paths.
26    """
27    schema_files = {
28        p.name.removesuffix(".sql")
29        for p in resource_path("schema").iterdir()
30        if p.is_file() and p.name.endswith(".sql")
31    }
32    workload_files = {
33        p.name.removesuffix(".sql")
34        for p in resource_path("workload").iterdir()
35        if p.is_file() and p.name.endswith(".sql")
36    }
37
38    return sorted(schema_files.intersection(workload_files))

Determines a list of avilable scenarios based on the intersection of files located in both the schema and workload resource paths.

class Scenario: 
41class Scenario:
42    def __init__(self, value: str) -> None:
43        self.value = value
44
45    def schema_path(self) -> Path:
46        return resource_path(f"schema/{self}.sql")
47
48    def workload_path(self) -> Path:
49        return resource_path(f"workload/{self}.sql")
50
51    def __str__(self) -> str:
52        return self.value
Scenario(value: str) 
42    def __init__(self, value: str) -> None:
43        self.value = value
value
def schema_path(self) -> pathlib.Path: 
45    def schema_path(self) -> Path:
46        return resource_path(f"schema/{self}.sql")
def workload_path(self) -> pathlib.Path: 
48    def workload_path(self) -> Path:
49        return resource_path(f"workload/{self}.sql")