Skip to content

App

materia.app

app

Context

Bases: TypedDict

config instance-attribute
config
logger instance-attribute
logger
database instance-attribute
database
cache instance-attribute
cache

ApplicationError

Bases: Exception

Application

Application(config)
PARAMETER DESCRIPTION
config

TYPE: Config

Source code in src/materia/app/app.py
36
37
38
39
40
41
42
43
44
45
46
47
def __init__(
    self,
    config: Config,
):
    self.config: Config = config
    self.logger: Optional[LoggerInstance] = None
    self.database: Optional[Database] = None
    self.cache: Optional[Cache] = None
    self.cron: Optional[Cron] = None
    self.backend: Optional[FastAPI] = None

    self.prepare_logger()
config instance-attribute
config = config
logger instance-attribute
logger = None
database instance-attribute
database = None
cache instance-attribute
cache = None
cron instance-attribute
cron = None
backend instance-attribute
backend = None
new async staticmethod
new(config)
PARAMETER DESCRIPTION
config

TYPE: Config

Source code in src/materia/app/app.py
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
@staticmethod
async def new(config: Config):
    app = Application(config)

    # if user := config.application.user:
    #    os.setuid(pwd.getpwnam(user).pw_uid)
    # if group := config.application.group:
    #    os.setgid(pwd.getpwnam(user).pw_gid)
    app.logger.debug("Initializing application...")
    await app.prepare_working_directory()

    try:
        await app.prepare_database()
        await app.prepare_cache()
        await app.prepare_cron()
        app.prepare_server()
    except Exception as e:
        app.logger.error(" ".join(e.args))
        sys.exit()

    try:
        import materia_frontend
    except ModuleNotFoundError:
        app.logger.warning(
            "`materia_frontend` is not installed. No user interface will be served."
        )

    return app
prepare_logger
prepare_logger()
Source code in src/materia/app/app.py
78
79
def prepare_logger(self):
    self.logger = Logger.new(**self.config.log.model_dump())
prepare_working_directory async
prepare_working_directory()
Source code in src/materia/app/app.py
81
82
83
84
85
86
87
88
async def prepare_working_directory(self):
    try:
        path = self.config.application.working_directory.resolve()
        self.logger.debug(f"Changing working directory to {path}")
        os.chdir(path)
    except FileNotFoundError as e:
        self.logger.error("Failed to change working directory: {}", e)
        sys.exit()
prepare_database async
prepare_database()
Source code in src/materia/app/app.py
90
91
92
93
async def prepare_database(self):
    url = self.config.database.url()
    self.logger.info("Connecting to database {}", url)
    self.database = await Database.new(url)  # type: ignore
prepare_cache async
prepare_cache()
Source code in src/materia/app/app.py
95
96
97
98
async def prepare_cache(self):
    url = self.config.cache.url()
    self.logger.info("Connecting to cache server {}", url)
    self.cache = await Cache.new(url)  # type: ignore
prepare_cron async
prepare_cron()
Source code in src/materia/app/app.py
100
101
102
103
104
105
async def prepare_cron(self):
    url = self.config.cache.url()
    self.logger.info("Prepairing cron")
    self.cron = Cron.new(
        self.config.cron.workers_count, backend_url=url, broker_url=url
    )
prepare_server
prepare_server()
Source code in src/materia/app/app.py
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
def prepare_server(self):
    @asynccontextmanager
    async def lifespan(app: FastAPI) -> AsyncIterator[Context]:
        yield Context(
            config=self.config,
            logger=self.logger,
            database=self.database,
            cache=self.cache,
        )

        if self.database.engine is not None:
            await self.database.dispose()

    self.backend = FastAPI(
        title="materia",
        version="0.1.0",
        docs_url=None,
        redoc_url=None,
        swagger_ui_init_oauth=None,
        swagger_ui_oauth2_redirect_url=None,
        openapi_url="/api/openapi.json",
        lifespan=lifespan,
    )
    self.backend.add_middleware(
        CORSMiddleware,
        allow_origins=["http://localhost", "http://localhost:5173"],
        allow_credentials=True,
        allow_methods=["*"],
        allow_headers=["*"],
    )
    self.backend.include_router(routers.docs.router)
    self.backend.include_router(routers.api.router)
    self.backend.include_router(routers.resources.router)
    self.backend.include_router(routers.root.router)

    for route in self.backend.routes:
        if isinstance(route, APIRoute):
            route.operation_id = (
                optional_string(optional(route.tags.__getitem__, 0), "{}_")
                + route.name
            )
start async
start()
Source code in src/materia/app/app.py
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
async def start(self):
    self.logger.info(f"Spinning up cron workers [{self.config.cron.workers_count}]")
    self.cron.run_workers()

    try:
        self.logger.info("Running database migrations")
        await self.database.run_migrations()

        uvicorn_config = uvicorn.Config(
            self.backend,
            port=self.config.server.port,
            host=str(self.config.server.address),
            log_config=Logger.uvicorn_config(self.config.log.level),
        )
        server = uvicorn.Server(uvicorn_config)

        await server.serve()
    except (KeyboardInterrupt, SystemExit):
        self.logger.info("Exiting...")
        sys.exit()
    except Exception as e:
        self.logger.error(" ".join(e.args))
        sys.exit()

asgi

MateriaWorker

Bases: UvicornWorker

CONFIG_KWARGS class-attribute instance-attribute
CONFIG_KWARGS = {
    "loop": "uvloop",
    "log_config": uvicorn_log_config(open(resolve())),
}

cli

cli

cli()
Source code in src/materia/app/cli.py
11
12
13
@click.group()
def cli():
    pass

start

start(config, debug)
PARAMETER DESCRIPTION
config

TYPE: Path

debug

TYPE: bool

Source code in src/materia/app/cli.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
@cli.command()
@click.option("--config", type=Path)
@click.option("--debug", "-d", is_flag=True, default=False, help="Enable debug output.")
def start(config: Path, debug: bool):
    config_path = config
    logger = Logger.new()

    # check the configuration file or use default
    if config_path is not None:
        config_path = config_path.resolve()
        try:
            logger.debug("Reading configuration file at {}", config_path)
            if not config_path.exists():
                logger.error("Configuration file was not found at {}.", config_path)
                sys.exit(1)
            else:
                config = Config.open(config_path.resolve())
        except Exception as e:
            logger.error("Failed to read configuration file: {}", e)
            sys.exit(1)
    else:
        # trying to find configuration file in the current working directory
        config_path = Config.data_dir().joinpath("config.toml")
        if config_path.exists():
            logger.info("Found configuration file in the current working directory.")
            try:
                config = Config.open(config_path)
            except Exception as e:
                logger.error("Failed to read configuration file: {}", e)
            else:
                logger.info("Using the default configuration.")
                config = Config()
        else:
            logger.info("Using the default configuration.")
            config = Config()

    if debug:
        config.log.level = "debug"

    async def main():
        app = await Application.new(config)
        await app.start()

    asyncio.run(main())

config

config()
Source code in src/materia/app/cli.py
62
63
64
@cli.group()
def config():
    pass

config_create

config_create(path, force)
PARAMETER DESCRIPTION
path

TYPE: Path

force

TYPE: bool

Source code in src/materia/app/cli.py
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
@config.command("create", help="Create a new configuration file.")
@click.option(
    "--path",
    "-p",
    type=Path,
    default=Path.cwd().joinpath("config.toml"),
    help="Path to the file.",
)
@click.option(
    "--force", "-f", is_flag=True, default=False, help="Overwrite a file if exists."
)
def config_create(path: Path, force: bool):
    path = path.resolve()
    config = Config()
    logger = Logger.new()

    if path.exists() and not force:
        logger.warning("File already exists at the given path. Exit.")
        sys.exit(1)

    if not path.parent.exists():
        logger.info("Creating directory at {}", path)
        path.parent.mkdir(parents=True)

    logger.info("Writing configuration file at {}", path)
    config.write(path)
    logger.info("All done.")

config_check

config_check(path)
PARAMETER DESCRIPTION
path

TYPE: Path

Source code in src/materia/app/cli.py
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
@config.command("check", help="Check the configuration file.")
@click.option(
    "--path",
    "-p",
    type=Path,
    default=Path.cwd().joinpath("config.toml"),
    help="Path to the file.",
)
def config_check(path: Path):
    path = path.resolve()
    logger = Logger.new()

    if not path.exists():
        logger.error("Configuration file was not found at the given path. Exit.")
        sys.exit(1)

    try:
        Config.open(path)
    except Exception as e:
        logger.error("{}", e)
    else:
        logger.info("OK.")

export

export()
Source code in src/materia/app/cli.py
120
121
122
@cli.group()
def export():
    pass

export_openapi

export_openapi(path)
PARAMETER DESCRIPTION
path

TYPE: Path

Source code in src/materia/app/cli.py
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
@export.command("openapi", help="Export an OpenAPI specification.")
@click.option(
    "--path",
    "-p",
    type=Path,
    default=Path.cwd().joinpath("openapi.json"),
    help="Path to the file.",
)
def export_openapi(path: Path):
    path = path.resolve()
    logger = Logger.new()
    config = Config()
    app = Application(config)
    app.prepare_server()

    logger.info("Writing file at {}", path)

    try:
        with open(path, "w") as io:
            json.dump(app.backend.openapi(), io, sort_keys=False)
    except Exception as e:
        logger.error("{}", e)

    logger.info("All done.")

wsgi

MateriaProcessManager

MateriaProcessManager(app, options=None)

Bases: WSGIApplication

PARAMETER DESCRIPTION
app

TYPE: str

options

TYPE: dict | None DEFAULT: None

Source code in src/materia/app/wsgi.py
6
7
8
9
def __init__(self, app: str, options: dict | None = None):
    self.app_uri = app 
    self.options = options or {}
    super().__init__()
app_uri instance-attribute
app_uri = app
options instance-attribute
options = options or {}
load_config
load_config()
Source code in src/materia/app/wsgi.py
11
12
13
14
15
16
17
18
def load_config(self):
    config = {
        key: value
        for key, value in self.options.items()
        if key in self.cfg.settings and value is not None
    }
    for key, value in config.items():
        self.cfg.set(key.lower(), value)

run

run()
Source code in src/materia/app/wsgi.py
20
21
22
23
24
25
26
27
28
29
def run():
    options = {
        "bind": "0.0.0.0:8000",
        "workers": (multiprocessing.cpu_count() * 2) + 1,
        "worker_class": "materia.app.wsgi.MateriaWorker",
        "raw_env": ["FOO=1"],
        "user": None,
        "group": None
    }
    MateriaProcessManager("materia.app.app:run", options).run()