Skip to content

Routers

materia.routers

api

router module-attribute

router = APIRouter(prefix='/api')

not_found

not_found()
Source code in src/materia/routers/api/__init__.py
15
16
17
@router.get("/api/{catchall:path}", status_code=404, include_in_schema=False)
def not_found():
    raise HTTPException(status_code=404)

auth

auth
router module-attribute
router = APIRouter(tags=['auth'])
signup async
signup(body, ctx=Depends())
PARAMETER DESCRIPTION
body

TYPE: UserCredentials

ctx

TYPE: Context DEFAULT: Depends()

Source code in src/materia/routers/api/auth/auth.py
11
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
42
43
44
@router.post("/auth/signup")
async def signup(body: UserCredentials, ctx: Context = Depends()):
    if not User.check_username(body.name):
        raise HTTPException(
            status.HTTP_500_INTERNAL_SERVER_ERROR, detail="Invalid username"
        )
    if not User.check_password(body.password, ctx.config):
        raise HTTPException(
            status.HTTP_500_INTERNAL_SERVER_ERROR,
            detail=f"Password is too short (minimum length {ctx.config.security.password_min_length})",
        )

    async with ctx.database.session() as session:
        if await User.by_name(body.name, session, with_lower=True):
            raise HTTPException(status.HTTP_409_CONFLICT, detail="User already exists")
        if await User.by_email(body.email, session):  # type: ignore
            raise HTTPException(status.HTTP_409_CONFLICT, detail="Email already used")

        count: Optional[int] = await User.count(session)

        await User(
            name=body.name,
            lower_name=body.name.lower(),
            full_name=body.name,
            email=body.email,
            hashed_password=security.hash_password(
                body.password, algo=ctx.config.security.password_hash_algo
            ),
            login_type=LoginType.Plain,
            # first registered user is admin
            is_admin=count == 0,
        ).new(session, ctx.config)

        await session.commit()
signin async
signin(body, response, ctx=Depends())
PARAMETER DESCRIPTION
body

TYPE: UserCredentials

response

TYPE: Response

ctx

TYPE: Context DEFAULT: Depends()

Source code in src/materia/routers/api/auth/auth.py
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
@router.post("/auth/signin")
async def signin(body: UserCredentials, response: Response, ctx: Context = Depends()):
    async with ctx.database.session() as session:
        if (current_user := await User.by_name(body.name, session)) is None:
            if (current_user := await User.by_email(str(body.email), session)) is None:
                raise HTTPException(
                    status.HTTP_401_UNAUTHORIZED, detail="Invalid email"
                )

    if not security.validate_password(
        body.password,
        current_user.hashed_password,
        algo=ctx.config.security.password_hash_algo,
    ):
        raise HTTPException(status.HTTP_401_UNAUTHORIZED, detail="Invalid password")

    issuer = "{}://{}".format(ctx.config.server.scheme, ctx.config.server.domain)
    secret = (
        ctx.config.oauth2.jwt_secret
        if ctx.config.oauth2.jwt_signing_algo in ["HS256", "HS384", "HS512"]
        else ctx.config.oauth2.jwt_signing_key
    )
    access_token = security.generate_token(
        str(current_user.id),
        str(secret),
        ctx.config.oauth2.access_token_lifetime,
        issuer,
    )
    refresh_token = security.generate_token(
        "", str(secret), ctx.config.oauth2.refresh_token_lifetime, issuer
    )

    response.set_cookie(
        ctx.config.security.cookie_access_token_name,
        value=access_token,
        max_age=ctx.config.oauth2.access_token_lifetime,
        secure=True,
        httponly=ctx.config.security.cookie_http_only,
        samesite="lax",
    )
    response.set_cookie(
        ctx.config.security.cookie_refresh_token_name,
        value=refresh_token,
        max_age=ctx.config.oauth2.refresh_token_lifetime,
        secure=True,
        httponly=ctx.config.security.cookie_http_only,
        samesite="lax",
    )
signout async
signout(response, ctx=Depends())
PARAMETER DESCRIPTION
response

TYPE: Response

ctx

TYPE: Context DEFAULT: Depends()

Source code in src/materia/routers/api/auth/auth.py
 97
 98
 99
100
@router.get("/auth/signout")
async def signout(response: Response, ctx: Context = Depends()):
    response.delete_cookie(ctx.config.security.cookie_access_token_name)
    response.delete_cookie(ctx.config.security.cookie_refresh_token_name)
oauth
router module-attribute
router = APIRouter(tags=['oauth2'])
OAuth2AuthorizationCodeRequestForm
OAuth2AuthorizationCodeRequestForm(
    redirect_uri,
    client_id,
    scope=None,
    state=None,
    response_type="code",
    grant_type="authorization_code",
)
PARAMETER DESCRIPTION
redirect_uri

TYPE: HttpUrl

client_id

TYPE: str

scope

TYPE: Union[str, None] DEFAULT: None

state

TYPE: Union[str, None] DEFAULT: None

response_type

TYPE: str DEFAULT: 'code'

grant_type

TYPE: str DEFAULT: 'authorization_code'

Source code in src/materia/routers/api/auth/oauth.py
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
def __init__(
    self, 
    redirect_uri: Annotated[HttpUrl, Form()],
    client_id: Annotated[str, Form()],
    scope: Annotated[Union[str, None], Form()] = None,
    state: Annotated[Union[str, None], Form()] = None,
    response_type: Annotated[str, Form()] = "code",
    grant_type: Annotated[str, Form(pattern = "password")] = "authorization_code"
) -> None:
    self.redirect_uri = redirect_uri 
    self.client_id = client_id
    self.scope = scope
    self.state = state
    self.response_type = response_type
    self.grant_type = grant_type
redirect_uri instance-attribute
redirect_uri = redirect_uri
client_id instance-attribute
client_id = client_id
scope instance-attribute
scope = scope
state instance-attribute
state = state
response_type instance-attribute
response_type = response_type
grant_type instance-attribute
grant_type = grant_type
AuthorizationCodeResponse

Bases: BaseModel

code instance-attribute
code
AccessTokenResponse

Bases: BaseModel

access_token instance-attribute
access_token
token_type instance-attribute
token_type
expires_in instance-attribute
expires_in
refresh_token instance-attribute
refresh_token
scope instance-attribute
scope
authorize async
authorize(form, ctx=Depends())
PARAMETER DESCRIPTION
form

TYPE: OAuth2AuthorizationCodeRequestForm

ctx

TYPE: Context DEFAULT: Depends()

Source code in src/materia/routers/api/auth/oauth.py
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
@router.post("/oauth2/authorize")
async def authorize(form: Annotated[OAuth2AuthorizationCodeRequestForm, Depends()], ctx: Context = Depends()):
    # grant_type: authorization_code, password_credentials, client_credentials, authorization_code (pkce)
    ctx.logger.debug(form)

    if form.grant_type == "authorization_code":
        # TODO: form validation 

        if not (app := await OAuth2Application.by_client_id(form.client_id, ctx.database)):
            raise HTTPException(status_code = HTTP_500_INTERNAL_SERVER_ERROR, detail = "Client ID not registered")

        if not (owner := await User.by_id(app.user_id, ctx.database)):
            raise HTTPException(status_code = HTTP_500_INTERNAL_SERVER_ERROR, detail = "User not found")

        if not app.contains_redirect_uri(form.redirect_uri):
            raise HTTPException(status_code = HTTP_500_INTERNAL_SERVER_ERROR, detail = "Unregistered redirect URI")

        if not form.response_type == "code":
            raise HTTPException(status_code = HTTP_500_INTERNAL_SERVER_ERROR, detail = "Unsupported response type")

        # TODO: code challenge (S256, plain, ...)
        # None: if not app.confidential_client: raise ...

        grant = await app.grant_by_user_id(owner.id, ctx.database)

        if app.confidential_client and grant is not None:
            code = await grant.generate_authorization_code(form.redirect_uri, ctx.cache)
            # TODO: include state to redirect_uri

            # return redirect 

        # redirect to grant page
    else:
        raise HTTPException(status_code = HTTP_500_INTERNAL_SERVER_ERROR, detail = "Unsupported grant type")

    pass 
token async
token(ctx=Depends())
PARAMETER DESCRIPTION
ctx

TYPE: Context DEFAULT: Depends()

Source code in src/materia/routers/api/auth/oauth.py
80
81
82
@router.post("/oauth2/access_token")
async def token(ctx: Context = Depends()):
    pass

directory

router module-attribute
router = APIRouter(tags=['directory'])
validate_current_directory async
validate_current_directory(
    path, repository, session, config
)
PARAMETER DESCRIPTION
path

TYPE: Path

repository

TYPE: Repository

session

TYPE: SessionContext

config

TYPE: Config

Source code in src/materia/routers/api/directory.py
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
async def validate_current_directory(
    path: Path, repository: Repository, session: SessionContext, config: Config
) -> Directory:
    if not FileSystem.check_path(path):
        raise HTTPException(status.HTTP_500_INTERNAL_SERVER_ERROR, "Invalid path")

    if not (
        directory := await Directory.by_path(
            repository,
            FileSystem.normalize(path),
            session,
            config,
        )
    ):
        raise HTTPException(status.HTTP_404_NOT_FOUND, "Directory not found")

    return directory
validate_target_directory async
validate_target_directory(
    path, repository, session, config
)
PARAMETER DESCRIPTION
path

TYPE: Path

repository

TYPE: Repository

session

TYPE: SessionContext

config

TYPE: Config

Source code in src/materia/routers/api/directory.py
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
async def validate_target_directory(
    path: Path, repository: Repository, session: SessionContext, config: Config
) -> Directory:
    if not FileSystem.check_path(path):
        raise HTTPException(
            status.HTTP_500_INTERNAL_SERVER_ERROR, "Invalid target path"
        )

    if FileSystem.normalize(path) == Path():
        # mean repository root
        target_directory = None
    else:
        if not (
            target_directory := await Directory.by_path(
                repository,
                FileSystem.normalize(path),
                session,
                config,
            )
        ):
            raise HTTPException(status.HTTP_404_NOT_FOUND, "Target directory not found")

    return target_directory
create async
create(
    path,
    repository=Depends(middleware.repository),
    ctx=Depends(),
)
PARAMETER DESCRIPTION
path

TYPE: DirectoryPath

repository

TYPE: Repository DEFAULT: Depends(repository)

ctx

TYPE: Context DEFAULT: Depends()

Source code in src/materia/routers/api/directory.py
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
@router.post("/directory")
async def create(
    path: DirectoryPath,
    repository: Repository = Depends(middleware.repository),
    ctx: middleware.Context = Depends(),
):
    if not FileSystem.check_path(path.path):
        raise HTTPException(status.HTTP_500_INTERNAL_SERVER_ERROR, "Invalid path")

    async with ctx.database.session() as session:
        current_directory = None
        current_path = Path()
        directory = None

        for part in FileSystem.normalize(path.path).parts:
            if not (
                directory := await Directory.by_path(
                    repository, current_path.joinpath(part), session, ctx.config
                )
            ):
                directory = await Directory(
                    repository_id=repository.id,
                    parent_id=current_directory.id if current_directory else None,
                    name=part,
                ).new(session, ctx.config)

            current_directory = directory
            current_path /= part

        await session.commit()
info async
info(
    path,
    repository=Depends(middleware.repository),
    ctx=Depends(),
)
PARAMETER DESCRIPTION
path

TYPE: Path

repository

TYPE: Repository DEFAULT: Depends(repository)

ctx

TYPE: Context DEFAULT: Depends()

Source code in src/materia/routers/api/directory.py
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
@router.get("/directory")
async def info(
    path: Path,
    repository: Repository = Depends(middleware.repository),
    ctx: middleware.Context = Depends(),
):
    async with ctx.database.session() as session:
        directory = await validate_current_directory(
            path, repository, session, ctx.config
        )

        info = await directory.info(session)

        return info
remove async
remove(
    path,
    repository=Depends(middleware.repository),
    ctx=Depends(),
)
PARAMETER DESCRIPTION
path

TYPE: Path

repository

TYPE: Repository DEFAULT: Depends(repository)

ctx

TYPE: Context DEFAULT: Depends()

Source code in src/materia/routers/api/directory.py
111
112
113
114
115
116
117
118
119
120
121
122
123
@router.delete("/directory")
async def remove(
    path: Path,
    repository: Repository = Depends(middleware.repository),
    ctx: middleware.Context = Depends(),
):
    async with ctx.database.session() as session:
        directory = await validate_current_directory(
            path, repository, session, ctx.config
        )

        await directory.remove(session, ctx.config)
        await session.commit()
rename async
rename(
    data,
    repository=Depends(middleware.repository),
    ctx=Depends(),
)
PARAMETER DESCRIPTION
data

TYPE: DirectoryRename

repository

TYPE: Repository DEFAULT: Depends(repository)

ctx

TYPE: Context DEFAULT: Depends()

Source code in src/materia/routers/api/directory.py
126
127
128
129
130
131
132
133
134
135
136
137
138
@router.patch("/directory/rename")
async def rename(
    data: DirectoryRename,
    repository: Repository = Depends(middleware.repository),
    ctx: middleware.Context = Depends(),
):
    async with ctx.database.session() as session:
        directory = await validate_current_directory(
            data.path, repository, session, ctx.config
        )

        await directory.rename(data.name, session, ctx.config, force=data.force)
        await session.commit()
move async
move(
    data,
    repository=Depends(middleware.repository),
    ctx=Depends(),
)
PARAMETER DESCRIPTION
data

TYPE: DirectoryCopyMove

repository

TYPE: Repository DEFAULT: Depends(repository)

ctx

TYPE: Context DEFAULT: Depends()

Source code in src/materia/routers/api/directory.py
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
@router.patch("/directory/move")
async def move(
    data: DirectoryCopyMove,
    repository: Repository = Depends(middleware.repository),
    ctx: middleware.Context = Depends(),
):
    async with ctx.database.session() as session:
        directory = await validate_current_directory(
            data.path, repository, session, ctx.config
        )
        target_directory = await validate_target_directory(
            data.target, repository, session, ctx.config
        )

        await directory.move(target_directory, session, ctx.config, force=data.force)
        await session.commit()
copy async
copy(
    data,
    repository=Depends(middleware.repository),
    ctx=Depends(),
)
PARAMETER DESCRIPTION
data

TYPE: DirectoryCopyMove

repository

TYPE: Repository DEFAULT: Depends(repository)

ctx

TYPE: Context DEFAULT: Depends()

Source code in src/materia/routers/api/directory.py
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
@router.post("/directory/copy")
async def copy(
    data: DirectoryCopyMove,
    repository: Repository = Depends(middleware.repository),
    ctx: middleware.Context = Depends(),
):
    async with ctx.database.session() as session:
        directory = await validate_current_directory(
            data.path, repository, session, ctx.config
        )
        target_directory = await validate_target_directory(
            data.target, repository, session, ctx.config
        )

        await directory.copy(target_directory, session, ctx.config, force=data.force)
        await session.commit()
content async
content(
    path,
    repository=Depends(middleware.repository),
    ctx=Depends(),
)
PARAMETER DESCRIPTION
path

TYPE: Path

repository

TYPE: Repository DEFAULT: Depends(repository)

ctx

TYPE: Context DEFAULT: Depends()

Source code in src/materia/routers/api/directory.py
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
@router.get("/directory/content", response_model=DirectoryContent)
async def content(
    path: Path,
    repository: Repository = Depends(middleware.repository),
    ctx: middleware.Context = Depends(),
):
    async with ctx.database.session() as session:
        directory = await validate_current_directory(
            path, repository, session, ctx.config
        )
        session.add(directory)
        await session.refresh(directory, attribute_names=["directories"])
        await session.refresh(directory, attribute_names=["files"])

        content = DirectoryContent(
            files=[await _file.info(session) for _file in directory.files],
            directories=[
                await _directory.info(session) for _directory in directory.directories
            ],
        )

    return content

docs

router module-attribute
router = APIRouter()
rapidoc async
rapidoc(request)
PARAMETER DESCRIPTION
request

TYPE: Request

Source code in src/materia/routers/api/docs.py
 7
 8
 9
10
11
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
@router.get("/docs", response_class=HTMLResponse, include_in_schema=False)
async def rapidoc(request: Request):
    return f"""
        <!doctype html>
        <html>
            <head>
                <meta charset="utf-8">
                <link href='http://fonts.googleapis.com/css?family=Roboto' rel='stylesheet' type='text/css'>
                <script 
                    type="module" 
                    src="https://unpkg.com/rapidoc/dist/rapidoc-min.js"
                ></script>
            </head>
            <body>
                <rapi-doc 
                    spec-url="{request.app.openapi_url}"
                    theme = "dark"
                    show-header = "false"
                    show-info = "true"
                    allow-authentication = "true"
                    allow-server-selection = "true"
                    allow-api-list-style-selection = "true"
                    theme = "dark"
                    render-style = "focused"
                    bg-color="#1e2129"
                    primary-color="#a47bea"
                    regular-font="Roboto"
                    mono-font="Roboto Mono"
                    show-method-in-nav-bar="as-colored-text">
                    <img slot="logo" style="display: none"/>
                </rapi-doc>
            </body> 
        </html>
    """

file

router module-attribute
router = APIRouter(tags=['file'])
FileSizeValidator
FileSizeValidator(capacity)
PARAMETER DESCRIPTION
capacity

TYPE: int

Source code in src/materia/routers/api/file.py
61
62
63
def __init__(self, capacity: int):
    self.body = 0
    self.capacity = capacity
body instance-attribute
body = 0
capacity instance-attribute
capacity = capacity
validate_current_file async
validate_current_file(path, repository, session, config)
PARAMETER DESCRIPTION
path

TYPE: Path

repository

TYPE: Repository

session

TYPE: SessionContext

config

TYPE: Config

Source code in src/materia/routers/api/file.py
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
async def validate_current_file(
    path: Path, repository: Repository, session: SessionContext, config: Config
) -> Directory:
    if not FileSystem.check_path(path):
        raise HTTPException(status.HTTP_500_INTERNAL_SERVER_ERROR, "Invalid path")

    if not (
        file := await File.by_path(
            repository,
            FileSystem.normalize(path),
            session,
            config,
        )
    ):
        raise HTTPException(status.HTTP_404_NOT_FOUND, "File not found")

    return file
create async
create(
    request,
    repository=Depends(middleware.repository),
    ctx=Depends(),
)
PARAMETER DESCRIPTION
request

TYPE: Request

repository

TYPE: Repository DEFAULT: Depends(repository)

ctx

TYPE: Context DEFAULT: Depends()

Source code in src/materia/routers/api/file.py
 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
@router.post("/file", openapi_extra={
    "requestBody" : {
        "content": {
            "multipart/form-data": {
                "schema": {
                    "required": ["file", "path"],
                    "type": "object",
                    "properties": {
                        "file": { "type": "string", "format": "binary" },
                        "path": { "type": "string", "format": "path", "example": "/"}
                    }
                }
            }
        },
        "required": True
    }
})
async def create(
    request: Request,
    repository: Repository = Depends(middleware.repository),
    ctx: middleware.Context = Depends(),
):
    async with ctx.database.session() as session:
        capacity = await repository.remaining_capacity(session)

    try:
        file = TemporaryFileTarget(
            ctx.config.application.working_directory,
            validator=FileSizeValidator(capacity),
        )
        path = ValueTarget()

        ctx.logger.debug(f"Shedule remove cache file: {file.path().name}")
        remove_cache_file.apply_async(args=(file.path(), ctx.config), countdown=10)

        parser = StreamingFormDataParser(headers=request.headers)
        parser.register("file", file)
        parser.register("path", path)

        async for chunk in request.stream():
            parser.data_received(chunk)

    except ClientDisconnect:
        file.remove()
        raise HTTPException(status.HTTP_500_INTERNAL_SERVER_ERROR, "Client disconnect")
    except HTTPException as e:
        file.remove()
        raise e
    except Exception as e:
        file.remove()
        raise HTTPException(status.HTTP_500_INTERNAL_SERVER_ERROR, " ".join(e.args))

    path = Path(path.value.decode())

    if not file.multipart_filename:
        file.remove()
        raise HTTPException(
            status.HTTP_417_EXPECTATION_FAILED, "Cannot upload file without name"
        )
    if not FileSystem.check_path(path):
        file.remove()
        raise HTTPException(status.HTTP_500_INTERNAL_SERVER_ERROR, "Invalid path")

    async with ctx.database.session() as session:
        target_directory = await validate_target_directory(
            path, repository, session, ctx.config
        )

        try:
            await File(
                repository_id=repository.id,
                parent_id=target_directory.id if target_directory else None,
                name=file.multipart_filename,
                size=await async_path.getsize(file.path()),
            ).new(file.path(), session, ctx.config)
        except Exception:
            raise HTTPException(
                status.HTTP_500_INTERNAL_SERVER_ERROR, "Failed to create file"
            )
        else:
            await session.commit()
info async
info(
    path,
    repository=Depends(middleware.repository),
    ctx=Depends(),
)
PARAMETER DESCRIPTION
path

TYPE: Path

repository

TYPE: Repository DEFAULT: Depends(repository)

ctx

TYPE: Context DEFAULT: Depends()

Source code in src/materia/routers/api/file.py
154
155
156
157
158
159
160
161
162
163
164
165
@router.get("/file", response_model=FileInfo)
async def info(
    path: Path,
    repository: Repository = Depends(middleware.repository),
    ctx: middleware.Context = Depends(),
):
    async with ctx.database.session() as session:
        file = await validate_current_file(path, repository, session, ctx.config)

        info = file.info()

        return info
remove async
remove(
    path,
    repository=Depends(middleware.repository),
    ctx=Depends(),
)
PARAMETER DESCRIPTION
path

TYPE: Path

repository

TYPE: Repository DEFAULT: Depends(repository)

ctx

TYPE: Context DEFAULT: Depends()

Source code in src/materia/routers/api/file.py
168
169
170
171
172
173
174
175
176
177
178
@router.delete("/file")
async def remove(
    path: Path,
    repository: Repository = Depends(middleware.repository),
    ctx: middleware.Context = Depends(),
):
    async with ctx.database.session() as session:
        file = await validate_current_file(path, repository, session, ctx.config)

        await file.remove(session, ctx.config)
        await session.commit()
rename async
rename(
    data,
    repository=Depends(middleware.repository),
    ctx=Depends(),
)
PARAMETER DESCRIPTION
data

TYPE: FileRename

repository

TYPE: Repository DEFAULT: Depends(repository)

ctx

TYPE: Context DEFAULT: Depends()

Source code in src/materia/routers/api/file.py
181
182
183
184
185
186
187
188
189
190
191
@router.patch("/file/rename")
async def rename(
    data: FileRename,
    repository: Repository = Depends(middleware.repository),
    ctx: middleware.Context = Depends(),
):
    async with ctx.database.session() as session:
        file = await validate_current_file(data.path, repository, session, ctx.config)

        await file.rename(data.name, session, ctx.config, force=data.force)
        await session.commit()
move async
move(
    data,
    repository=Depends(middleware.repository),
    ctx=Depends(),
)
PARAMETER DESCRIPTION
data

TYPE: FileCopyMove

repository

TYPE: Repository DEFAULT: Depends(repository)

ctx

TYPE: Context DEFAULT: Depends()

Source code in src/materia/routers/api/file.py
194
195
196
197
198
199
200
201
202
203
204
205
206
207
@router.patch("/file/move")
async def move(
    data: FileCopyMove,
    repository: Repository = Depends(middleware.repository),
    ctx: middleware.Context = Depends(),
):
    async with ctx.database.session() as session:
        file = await validate_current_file(data.path, repository, session, ctx.config)
        target_directory = await validate_target_directory(
            data.target, repository, session, ctx.config
        )

        await file.move(target_directory, session, ctx.config, force=data.force)
        await session.commit()
copy async
copy(
    data,
    repository=Depends(middleware.repository),
    ctx=Depends(),
)
PARAMETER DESCRIPTION
data

TYPE: FileCopyMove

repository

TYPE: Repository DEFAULT: Depends(repository)

ctx

TYPE: Context DEFAULT: Depends()

Source code in src/materia/routers/api/file.py
210
211
212
213
214
215
216
217
218
219
220
221
222
223
@router.post("/file/copy")
async def copy(
    data: FileCopyMove,
    repository: Repository = Depends(middleware.repository),
    ctx: middleware.Context = Depends(),
):
    async with ctx.database.session() as session:
        file = await validate_current_file(data.path, repository, session, ctx.config)
        target_directory = await validate_target_directory(
            data.target, repository, session, ctx.config
        )

        await file.copy(target_directory, session, ctx.config, force=data.force)
        await session.commit()

repository

router module-attribute
router = APIRouter(tags=['repository'])
create async
create(user=Depends(middleware.user), ctx=Depends())
PARAMETER DESCRIPTION
user

TYPE: User DEFAULT: Depends(user)

ctx

TYPE: Context DEFAULT: Depends()

Source code in src/materia/routers/api/repository.py
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
@router.post("/repository")
async def create(
    user: User = Depends(middleware.user), ctx: middleware.Context = Depends()
):
    async with ctx.database.session() as session:
        if await Repository.from_user(user, session):
            raise HTTPException(status.HTTP_409_CONFLICT, "Repository already exists")

    async with ctx.database.session() as session:
        try:
            await Repository(
                user_id=user.id, capacity=ctx.config.repository.capacity
            ).new(session, ctx.config)
            await session.commit()
        except Exception as e:
            raise HTTPException(
                status.HTTP_500_INTERNAL_SERVER_ERROR, detail=" ".join(e.args)
            )
info async
info(
    repository=Depends(middleware.repository), ctx=Depends()
)
PARAMETER DESCRIPTION
repository

DEFAULT: Depends(repository)

ctx

TYPE: Context DEFAULT: Depends()

Source code in src/materia/routers/api/repository.py
36
37
38
39
40
41
@router.get("/repository", response_model=RepositoryInfo)
async def info(
    repository=Depends(middleware.repository), ctx: middleware.Context = Depends()
):
    async with ctx.database.session() as session:
        return await repository.info(session)
remove async
remove(
    repository=Depends(middleware.repository), ctx=Depends()
)
PARAMETER DESCRIPTION
repository

DEFAULT: Depends(repository)

ctx

TYPE: Context DEFAULT: Depends()

Source code in src/materia/routers/api/repository.py
44
45
46
47
48
49
50
51
52
53
54
@router.delete("/repository")
async def remove(
    repository=Depends(middleware.repository),
    ctx: middleware.Context = Depends(),
):
    try:
        async with ctx.database.session() as session:
            await repository.remove(session, ctx.config)
            await session.commit()
    except Exception as e:
        raise HTTPException(status.HTTP_500_INTERNAL_SERVER_ERROR, f"{e}")
content async
content(
    repository=Depends(middleware.repository), ctx=Depends()
)
PARAMETER DESCRIPTION
repository

DEFAULT: Depends(repository)

ctx

TYPE: Context DEFAULT: Depends()

Source code in src/materia/routers/api/repository.py
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
@router.get("/repository/content", response_model=RepositoryContent)
async def content(
    repository=Depends(middleware.repository), ctx: middleware.Context = Depends()
):
    async with ctx.database.session() as session:
        session.add(repository)
        await session.refresh(repository, attribute_names=["directories"])
        await session.refresh(repository, attribute_names=["files"])

        content = RepositoryContent(
            files=[await _file.info(session) for _file in repository.files],
            directories=[
                await _directory.info(session) for _directory in repository.directories
            ],
        )

    return content

tasks

router module-attribute
router = APIRouter(tags=['tasks'])
status_task async
status_task(task_id)
PARAMETER DESCRIPTION
task_id

Source code in src/materia/routers/api/tasks.py
 8
 9
10
11
12
13
14
15
16
@router.get("/tasks/${task_id}")
async def status_task(task_id):
    task_result = AsyncResult(task_id)
    result = {
        "task_id": task_id,
        "task_status": task_result.status,
        "task_result": task_result.result,
    }
    return JSONResponse(result)

user

router module-attribute
router = APIRouter(tags=['user'])
info async
info(claims=Depends(middleware.jwt_cookie), ctx=Depends())
PARAMETER DESCRIPTION
claims

DEFAULT: Depends(jwt_cookie)

ctx

TYPE: Context DEFAULT: Depends()

Source code in src/materia/routers/api/user.py
11
12
13
14
15
16
17
18
19
@router.get("/user", response_model=UserInfo)
async def info(
    claims=Depends(middleware.jwt_cookie), ctx: middleware.Context = Depends()
):
    async with ctx.database.session() as session:
        if not (current_user := await User.by_id(uuid.UUID(claims.sub), session)):
            raise HTTPException(status.HTTP_401_UNAUTHORIZED, "Missing user")

    return current_user.info()
remove async
remove(user=Depends(middleware.user), ctx=Depends())
PARAMETER DESCRIPTION
user

TYPE: User DEFAULT: Depends(user)

ctx

TYPE: Context DEFAULT: Depends()

Source code in src/materia/routers/api/user.py
22
23
24
25
26
27
28
29
30
31
32
33
34
@router.delete("/user")
async def remove(
    user: User = Depends(middleware.user), ctx: middleware.Context = Depends()
):
    try:
        async with ctx.database.session() as session:
            await user.remove(session)
            await session.commit()

    except Exception as e:
        raise HTTPException(
            status.HTTP_500_INTERNAL_SERVER_ERROR, f"Failed to remove user: {e}"
        ) from e
avatar async
avatar(file, user=Depends(middleware.user), ctx=Depends())
PARAMETER DESCRIPTION
file

TYPE: UploadFile

user

TYPE: User DEFAULT: Depends(user)

ctx

TYPE: Context DEFAULT: Depends()

Source code in src/materia/routers/api/user.py
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
@router.put("/user/avatar")
async def avatar(
    file: UploadFile,
    user: User = Depends(middleware.user),
    ctx: middleware.Context = Depends(),
):
    async with ctx.database.session() as session:
        try:
            await user.edit_avatar(io.BytesIO(await file.read()), session, ctx.config)
            await session.commit()
        except Exception as e:
            raise HTTPException(
                status.HTTP_500_INTERNAL_SERVER_ERROR,
                f"{e}",
            )
remove_avatar async
remove_avatar(user=Depends(middleware.user), ctx=Depends())
PARAMETER DESCRIPTION
user

TYPE: User DEFAULT: Depends(user)

ctx

TYPE: Context DEFAULT: Depends()

Source code in src/materia/routers/api/user.py
54
55
56
57
58
59
60
61
62
63
64
65
66
67
@router.delete("/user/avatar")
async def remove_avatar(
    user: User = Depends(middleware.user),
    ctx: middleware.Context = Depends(),
):
    async with ctx.database.session() as session:
        try:
            await user.edit_avatar(None, session, ctx.config)
            await session.commit()
        except Exception as e:
            raise HTTPException(
                status.HTTP_500_INTERNAL_SERVER_ERROR,
                f"{e}",
            )

docs

router module-attribute

router = APIRouter()

docs async

docs(request, ctx=Depends())
PARAMETER DESCRIPTION
request

TYPE: Request

ctx

TYPE: Context DEFAULT: Depends()

Source code in src/materia/routers/docs.py
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
@router.get("/docs/{catchall:path}", include_in_schema=False)
async def docs(request: Request, ctx: middleware.Context = Depends()):
    docs_directory = Path(materia_docs.__path__[0]).resolve()
    target = docs_directory.joinpath(request.path_params["catchall"]).resolve()

    if not optional(target.relative_to, docs_directory):
        raise HTTPException(status.HTTP_403_FORBIDDEN)

    if target.is_dir() and (index := target.joinpath("index.html")).is_file():
        return FileResponse(index)

    if not target.is_file():
        raise HTTPException(status.HTTP_404_NOT_FOUND)

    return FileResponse(target)

middleware

Context

Context(request)
PARAMETER DESCRIPTION
request

TYPE: Request

Source code in src/materia/routers/middleware.py
26
27
28
29
30
def __init__(self, request: Request):
    self.config = request.state.config
    self.database = request.state.database
    self.cache = request.state.cache
    self.logger = request.state.logger
config instance-attribute
config = config
database instance-attribute
database = database
cache instance-attribute
cache = cache
logger instance-attribute
logger = logger
jwt_cookie(request, response, ctx=Depends())
PARAMETER DESCRIPTION
request

TYPE: Request

response

TYPE: Response

ctx

TYPE: Context DEFAULT: Depends()

Source code in src/materia/routers/middleware.py
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
async def jwt_cookie(request: Request, response: Response, ctx: Context = Depends()):
    if not (
        access_token := request.cookies.get(
            ctx.config.security.cookie_access_token_name
        )
    ):
        raise HTTPException(status.HTTP_401_UNAUTHORIZED, "Missing token")
    refresh_token = request.cookies.get(ctx.config.security.cookie_refresh_token_name)

    if ctx.config.oauth2.jwt_signing_algo in ["HS256", "HS384", "HS512"]:
        secret = ctx.config.oauth2.jwt_secret
    else:
        secret = ctx.config.oauth2.jwt_signing_key

    issuer = "{}://{}".format(ctx.config.server.scheme, ctx.config.server.domain)

    try:
        refresh_claims = (
            security.validate_token(refresh_token, secret) if refresh_token else None
        )

        if refresh_claims:
            if refresh_claims.exp < datetime.now().timestamp():
                refresh_claims = None
    except jwt.PyJWTError:
        refresh_claims = None

    try:
        access_claims = security.validate_token(access_token, secret)

        if access_claims.exp < datetime.now().timestamp():
            if refresh_claims:
                new_access_token = security.generate_token(
                    access_claims.sub,
                    str(secret),
                    ctx.config.oauth2.access_token_lifetime,
                    issuer,
                )
                access_claims = security.validate_token(new_access_token, secret)
                response.set_cookie(
                    ctx.config.security.cookie_access_token_name,
                    value=new_access_token,
                    max_age=ctx.config.oauth2.access_token_lifetime,
                    secure=True,
                    httponly=ctx.config.security.cookie_http_only,
                    samesite="lax",
                )
            else:
                access_claims = None
    except jwt.PyJWTError as e:
        raise HTTPException(status.HTTP_401_UNAUTHORIZED, f"Invalid token: {e}")

    async with ctx.database.session() as session:
        if not await User.by_id(uuid.UUID(access_claims.sub), session):
            raise HTTPException(status.HTTP_401_UNAUTHORIZED, "Invalid user")

    return access_claims

user async

user(claims=Depends(jwt_cookie), ctx=Depends())
PARAMETER DESCRIPTION
claims

DEFAULT: Depends(jwt_cookie)

ctx

TYPE: Context DEFAULT: Depends()

Source code in src/materia/routers/middleware.py
92
93
94
95
96
97
async def user(claims=Depends(jwt_cookie), ctx: Context = Depends()) -> User:
    async with ctx.database.session() as session:
        if not (current_user := await User.by_id(uuid.UUID(claims.sub), session)):
            raise HTTPException(status.HTTP_401_UNAUTHORIZED, "Missing user")

    return current_user

repository async

repository(user=Depends(user), ctx=Depends())
PARAMETER DESCRIPTION
user

TYPE: User DEFAULT: Depends(user)

ctx

TYPE: Context DEFAULT: Depends()

Source code in src/materia/routers/middleware.py
100
101
102
103
104
105
106
107
108
async def repository(user: User = Depends(user), ctx: Context = Depends()):
    async with ctx.database.session() as session:
        session.add(user)
        await session.refresh(user, attribute_names=["repository"])

    if not (repository := user.repository):
        raise HTTPException(status.HTTP_404_NOT_FOUND, "Repository not found")

    return repository

repository_path async

repository_path(user=Depends(user), ctx=Depends())
PARAMETER DESCRIPTION
user

TYPE: User DEFAULT: Depends(user)

ctx

TYPE: Context DEFAULT: Depends()

Source code in src/materia/routers/middleware.py
111
112
async def repository_path(user: User = Depends(user), ctx: Context = Depends()) -> Path:
    return ctx.config.data_dir() / "repository" / user.lower_name

resources

router module-attribute

router = APIRouter(tags=['resources'], prefix='/resources')

avatar async

avatar(avatar_id, format='png', ctx=Depends())
PARAMETER DESCRIPTION
avatar_id

TYPE: str

format

TYPE: str DEFAULT: 'png'

ctx

TYPE: Context DEFAULT: Depends()

Source code in src/materia/routers/resources.py
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
@router.get("/avatars/{avatar_id}")
async def avatar(
    avatar_id: str, format: str = "png", ctx: middleware.Context = Depends()
):
    avatar_path = Config.data_dir() / "avatars" / avatar_id
    format = format.upper()

    if not avatar_path.exists():
        raise HTTPException(
            status.HTTP_404_NOT_FOUND, "Failed to find the given avatar"
        )

    try:
        img = Image.open(avatar_path)
        buffer = io.BytesIO()

        if format == "JPEG":
            img.convert("RGB")

        img.save(buffer, format=format)

    except OSError:
        raise HTTPException(
            status.HTTP_422_UNPROCESSABLE_ENTITY, "Failed to process image file"
        )

    return Response(content=buffer.getvalue(), media_type=Image.MIME[format])

assets async

assets(filename)
PARAMETER DESCRIPTION
filename

TYPE: str

Source code in src/materia/routers/resources.py
48
49
50
51
52
53
54
55
56
57
58
59
60
@router.get("/assets/{filename}")
async def assets(filename: str):
    path = Path(materia_frontend.__path__[0]).joinpath(
        "dist", "resources", "assets", filename
    )

    if not path.exists():
        return Response(status_code=status.HTTP_404_NOT_FOUND)

    content = path.read_bytes()
    mime = mimetypes.guess_type(path)[0]

    return Response(content, media_type=mime)

root

router module-attribute

router = APIRouter(tags=['root'])

templates module-attribute

templates = Jinja2Templates(
    directory=Path(__path__[0]) / "dist"
)

root async

root(request)
PARAMETER DESCRIPTION
request

TYPE: Request

Source code in src/materia/routers/root.py
16
17
18
19
@router.get("/{spa:path}", response_class=HTMLResponse, include_in_schema=False)
async def root(request: Request):
    # raise HTTPException(404)
    return templates.TemplateResponse(request, "base.html", {"view": "app"})