Skip to content

Models

materia.models

auth

access_token

oauth2

OAuth2Application

Bases: Base

id class-attribute instance-attribute
id = mapped_column(BigInteger, primary_key=True)
user_id class-attribute instance-attribute
user_id = mapped_column(
    ForeignKey("user.id", ondelete="CASCADE")
)
name instance-attribute
name
client_id class-attribute instance-attribute
client_id = mapped_column(default=uuid4)
hashed_client_secret instance-attribute
hashed_client_secret
redirect_uris class-attribute instance-attribute
redirect_uris = mapped_column(JSON)
confidential_client class-attribute instance-attribute
confidential_client = mapped_column(default=True)
created class-attribute instance-attribute
created = mapped_column(BigInteger, default=time)
updated class-attribute instance-attribute
updated = mapped_column(BigInteger, default=time)
grants class-attribute instance-attribute
grants = relationship(back_populates='application')
to_dict
to_dict()
Source code in src/materia/models/base.py
7
8
def to_dict(self) -> dict:
    return {key: getattr(self, key) for key in self.__table__.columns.keys()}
clone
clone()

Clone model. Included: columns and values, foreign keys Ignored: primary keys, relationships

Source code in src/materia/models/base.py
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
def clone(self) -> Optional[Self]:
    """Clone model.
    Included: columns and values, foreign keys
    Ignored: primary keys, relationships
    """
    # if not inspect(self).persistent:
    #    return

    cloned = self.__class__(
        **{
            key: getattr(self, key)
            for key in self.__table__.columns.keys()
            # ignore primary keys
            if key not in self.__table__.primary_key.columns.keys()
        }
    )

    return cloned
contains_redirect_uri
contains_redirect_uri(uri)
PARAMETER DESCRIPTION
uri

TYPE: HttpUrl

Source code in src/materia/models/auth/oauth2.py
32
33
34
35
36
37
38
39
40
41
def contains_redirect_uri(self, uri: HttpUrl) -> bool:
    if not self.confidential_client:
        if uri.scheme == "http" and uri.host in ["127.0.0.1", "[::1]"]:
            return uri in self.redirect_uris

    else:
        if uri.scheme == "https" and uri.port == 443:
            return uri in self.redirect_uris

    return False
generate_client_secret async
generate_client_secret(db)
PARAMETER DESCRIPTION
db

TYPE: Database

Source code in src/materia/models/auth/oauth2.py
43
44
45
46
47
48
49
50
51
52
53
async def generate_client_secret(self, db: Database) -> str:
    client_secret = security.generate_key()
    hashed_secret = bcrypt.hashpw(client_secret, bcrypt.gensalt())

    self.hashed_client_secret = str(hashed_secret)

    async with db.session() as session:
        session.add(self)
        await session.commit()

    return str(client_secret)
validate_client_secret
validate_client_secret(secret)
PARAMETER DESCRIPTION
secret

TYPE: bytes

Source code in src/materia/models/auth/oauth2.py
55
56
def validate_client_secret(self, secret: bytes) -> bool:
    return bcrypt.checkpw(secret, self.hashed_client_secret.encode())
update async staticmethod
update(db, app)
PARAMETER DESCRIPTION
db

TYPE: Database

app

TYPE: OAuth2Application

Source code in src/materia/models/auth/oauth2.py
58
59
60
61
62
@staticmethod
async def update(db: Database, app: "OAuth2Application"):
    async with db.session() as session:
        session.add(app)
        await session.commit()
delete async staticmethod
delete(db, id, user_id)
PARAMETER DESCRIPTION
db

TYPE: Database

id

TYPE: int

user_id

TYPE: int

Source code in src/materia/models/auth/oauth2.py
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
@staticmethod
async def delete(db: Database, id: int, user_id: int):
    async with db.session() as session:
        if not (
            application := (
                await session.scalars(
                    select(OAuth2Application).where(
                        and_(
                            OAuth2Application.id == id,
                            OAuth2Application.user_id == user_id,
                        )
                    )
                )
            ).first()
        ):
            raise Exception("OAuth2Application not found")

        # await session.refresh(application, attribute_names = [ "grants" ])
        await session.delete(application)
by_client_id async staticmethod
by_client_id(client_id, db)
PARAMETER DESCRIPTION
client_id

TYPE: str

db

TYPE: Database

Source code in src/materia/models/auth/oauth2.py
84
85
86
87
88
89
90
91
@staticmethod
async def by_client_id(client_id: str, db: Database) -> Union[Self, None]:
    async with db.session() as session:
        return await session.scalar(
            select(OAuth2Application).where(
                OAuth2Application.client_id == client_id
            )
        )
grant_by_user_id async
grant_by_user_id(user_id, db)
PARAMETER DESCRIPTION
user_id

TYPE: UUID

db

TYPE: Database

Source code in src/materia/models/auth/oauth2.py
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
async def grant_by_user_id(
    self, user_id: UUID, db: Database
) -> Union["OAuth2Grant", None]:
    async with db.session() as session:
        return (
            await session.scalars(
                select(OAuth2Grant).where(
                    and_(
                        OAuth2Grant.application_id == self.id,
                        OAuth2Grant.user_id == user_id,
                    )
                )
            )
        ).first()
OAuth2AuthorizationCode

Bases: BaseModel

grant instance-attribute
grant
code instance-attribute
code
redirect_uri instance-attribute
redirect_uri
created instance-attribute
created
lifetime instance-attribute
lifetime
generate_redirect_uri
generate_redirect_uri(state=None)
PARAMETER DESCRIPTION
state

TYPE: Optional[str] DEFAULT: None

Source code in src/materia/models/auth/oauth2.py
116
117
118
119
120
121
122
123
124
def generate_redirect_uri(self, state: Optional[str] = None) -> httpx.URL:
    redirect = httpx.URL(str(self.redirect_uri))

    if state:
        redirect = redirect.copy_add_param("state", state)

    redirect = redirect.copy_add_param("code", self.code)

    return redirect
OAuth2Grant

Bases: Base

id class-attribute instance-attribute
id = mapped_column(BigInteger, primary_key=True)
user_id class-attribute instance-attribute
user_id = mapped_column(
    ForeignKey("user.id", ondelete="CASCADE")
)
application_id class-attribute instance-attribute
application_id = mapped_column(
    ForeignKey("oauth2_application.id", ondelete="CASCADE")
)
scope instance-attribute
scope
created class-attribute instance-attribute
created = mapped_column(default=time)
updated class-attribute instance-attribute
updated = mapped_column(default=time)
application class-attribute instance-attribute
application = relationship(back_populates='grants')
to_dict
to_dict()
Source code in src/materia/models/base.py
7
8
def to_dict(self) -> dict:
    return {key: getattr(self, key) for key in self.__table__.columns.keys()}
clone
clone()

Clone model. Included: columns and values, foreign keys Ignored: primary keys, relationships

Source code in src/materia/models/base.py
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
def clone(self) -> Optional[Self]:
    """Clone model.
    Included: columns and values, foreign keys
    Ignored: primary keys, relationships
    """
    # if not inspect(self).persistent:
    #    return

    cloned = self.__class__(
        **{
            key: getattr(self, key)
            for key in self.__table__.columns.keys()
            # ignore primary keys
            if key not in self.__table__.primary_key.columns.keys()
        }
    )

    return cloned
generate_authorization_code async
generate_authorization_code(redirect_uri, cache)
PARAMETER DESCRIPTION
redirect_uri

TYPE: HttpUrl

cache

TYPE: Cache

Source code in src/materia/models/auth/oauth2.py
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
async def generate_authorization_code(
    self, redirect_uri: HttpUrl, cache: Cache
) -> OAuth2AuthorizationCode:
    code = OAuth2AuthorizationCode(
        grant=self,
        redirect_uri=redirect_uri,
        code=security.generate_key().decode(),
        created=int(time()),
        lifetime=3000,
    )

    async with cache.client() as client:
        client.set(
            "oauth2_authorization_code_{}".format(code.created),
            code.code,
            ex=code.lifetime,
        )

    return code
scope_contains
scope_contains(scope)
PARAMETER DESCRIPTION
scope

TYPE: str

Source code in src/materia/models/auth/oauth2.py
161
162
def scope_contains(self, scope: str) -> bool:
    return scope in self.scope.split(" ")

source

LoginType

Bases: Enum

Plain class-attribute instance-attribute
Plain = auto()
OAuth2 class-attribute instance-attribute
OAuth2 = auto()
Smtp class-attribute instance-attribute
Smtp = auto()
LoginSource

Bases: Base

id class-attribute instance-attribute
id = mapped_column(BigInteger, primary_key=True)
type instance-attribute
type
created class-attribute instance-attribute
created = mapped_column(default=time)
updated class-attribute instance-attribute
updated = mapped_column(default=time)
to_dict
to_dict()
Source code in src/materia/models/base.py
7
8
def to_dict(self) -> dict:
    return {key: getattr(self, key) for key in self.__table__.columns.keys()}
clone
clone()

Clone model. Included: columns and values, foreign keys Ignored: primary keys, relationships

Source code in src/materia/models/base.py
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
def clone(self) -> Optional[Self]:
    """Clone model.
    Included: columns and values, foreign keys
    Ignored: primary keys, relationships
    """
    # if not inspect(self).persistent:
    #    return

    cloned = self.__class__(
        **{
            key: getattr(self, key)
            for key in self.__table__.columns.keys()
            # ignore primary keys
            if key not in self.__table__.primary_key.columns.keys()
        }
    )

    return cloned
is_plain
is_plain()
Source code in src/materia/models/auth/source.py
24
25
def is_plain(self) -> bool:
    return self.type == LoginType.Plain
is_oauth2
is_oauth2()
Source code in src/materia/models/auth/source.py
27
28
def is_oauth2(self) -> bool:
    return self.type == LoginType.OAuth2
is_smtp
is_smtp()
Source code in src/materia/models/auth/source.py
30
31
def is_smtp(self) -> bool:
    return self.type == LoginType.Smtp

base

Base

Bases: DeclarativeBase

to_dict
to_dict()
Source code in src/materia/models/base.py
7
8
def to_dict(self) -> dict:
    return {key: getattr(self, key) for key in self.__table__.columns.keys()}
clone
clone()

Clone model. Included: columns and values, foreign keys Ignored: primary keys, relationships

Source code in src/materia/models/base.py
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
def clone(self) -> Optional[Self]:
    """Clone model.
    Included: columns and values, foreign keys
    Ignored: primary keys, relationships
    """
    # if not inspect(self).persistent:
    #    return

    cloned = self.__class__(
        **{
            key: getattr(self, key)
            for key in self.__table__.columns.keys()
            # ignore primary keys
            if key not in self.__table__.primary_key.columns.keys()
        }
    )

    return cloned

directory

DirectoryError

Bases: Exception

Directory

Bases: Base

id class-attribute instance-attribute
id = mapped_column(BigInteger, primary_key=True)
repository_id class-attribute instance-attribute
repository_id = mapped_column(
    ForeignKey("repository.id", ondelete="CASCADE")
)
parent_id class-attribute instance-attribute
parent_id = mapped_column(
    ForeignKey("directory.id", ondelete="CASCADE"),
    nullable=True,
)
created class-attribute instance-attribute
created = mapped_column(
    BigInteger, nullable=False, default=time
)
updated class-attribute instance-attribute
updated = mapped_column(
    BigInteger, nullable=False, default=time
)
name instance-attribute
name
is_public class-attribute instance-attribute
is_public = mapped_column(default=False)
repository class-attribute instance-attribute
repository = relationship(back_populates='directories')
directories class-attribute instance-attribute
directories = relationship(back_populates='parent')
parent class-attribute instance-attribute
parent = relationship(
    back_populates="directories", remote_side=[id]
)
files class-attribute instance-attribute
files = relationship(back_populates='parent')
link = relationship(back_populates='directory')
to_dict
to_dict()
Source code in src/materia/models/base.py
7
8
def to_dict(self) -> dict:
    return {key: getattr(self, key) for key in self.__table__.columns.keys()}
clone
clone()

Clone model. Included: columns and values, foreign keys Ignored: primary keys, relationships

Source code in src/materia/models/base.py
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
def clone(self) -> Optional[Self]:
    """Clone model.
    Included: columns and values, foreign keys
    Ignored: primary keys, relationships
    """
    # if not inspect(self).persistent:
    #    return

    cloned = self.__class__(
        **{
            key: getattr(self, key)
            for key in self.__table__.columns.keys()
            # ignore primary keys
            if key not in self.__table__.primary_key.columns.keys()
        }
    )

    return cloned
new async
new(session, config)
PARAMETER DESCRIPTION
session

TYPE: SessionContext

config

TYPE: Config

Source code in src/materia/models/directory.py
41
42
43
44
45
46
47
48
49
50
51
52
async def new(self, session: SessionContext, config: Config) -> Optional[Self]:
    session.add(self)
    await session.flush()
    await session.refresh(self, attribute_names=["repository"])

    repository_path = await self.repository.real_path(session, config)
    directory_path = await self.real_path(session, config)

    new_directory = FileSystem(directory_path, repository_path)
    await new_directory.make_directory()

    return self
remove async
remove(session, config)
PARAMETER DESCRIPTION
session

TYPE: SessionContext

config

TYPE: Config

Source code in src/materia/models/directory.py
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
async def remove(self, session: SessionContext, config: Config):
    session.add(self)
    await session.refresh(
        self, attribute_names=["repository", "directories", "files"]
    )

    if self.directories:
        for directory in self.directories:
            await directory.remove(session, config)

    if self.files:
        for file in self.files:
            await file.remove(session, config)

    repository_path = await self.repository.real_path(session, config)
    directory_path = await self.real_path(session, config)

    current_directory = FileSystem(directory_path, repository_path)
    await current_directory.remove()

    await session.delete(self)
    await session.flush()
relative_path async
relative_path(session)

Get path of the directory relative repository root.

PARAMETER DESCRIPTION
session

TYPE: SessionContext

Source code in src/materia/models/directory.py
 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
async def relative_path(self, session: SessionContext) -> Optional[Path]:
    """Get path of the directory relative repository root."""
    if inspect(self).was_deleted:
        return None

    parts = []
    current_directory = self

    while True:
        # ISSUE: accessing `parent` attribute raises greenlet_spawn has not been called; can't call await_only() here
        # parts.append(current_directory.name)
        # session.add(current_directory)
        # await session.refresh(current_directory, attribute_names=["parent"])
        # if current_directory.parent is None:
        #    break
        # current_directory = current_directory.parent

        parts.append(current_directory.name)

        if current_directory.parent_id is None:
            break

        current_directory = (
            await session.scalars(
                sa.select(Directory).where(
                    Directory.id == current_directory.parent_id,
                )
            )
        ).first()

    return Path().joinpath(*reversed(parts))
real_path async
real_path(session, config)

Get absolute path of the directory

PARAMETER DESCRIPTION
session

TYPE: SessionContext

config

TYPE: Config

Source code in src/materia/models/directory.py
109
110
111
112
113
114
115
116
117
118
119
async def real_path(
    self, session: SessionContext, config: Config
) -> Optional[Path]:
    """Get absolute path of the directory"""
    if inspect(self).was_deleted:
        return None

    repository_path = await self.repository.real_path(session, config)
    relative_path = await self.relative_path(session)

    return repository_path.joinpath(relative_path)
is_root
is_root()
Source code in src/materia/models/directory.py
121
122
def is_root(self) -> bool:
    return self.parent_id is None
by_path async staticmethod
by_path(repository, path, session, config)
PARAMETER DESCRIPTION
repository

TYPE: Repository

path

TYPE: Path

session

TYPE: SessionContext

config

TYPE: Config

Source code in src/materia/models/directory.py
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
@staticmethod
async def by_path(
    repository: "Repository", path: Path, session: SessionContext, config: Config
) -> Optional[Self]:
    if path == Path():
        raise DirectoryError("Cannot find directory by empty path")

    current_directory: Optional[Directory] = None

    for part in path.parts:
        # from root directory to target directory
        current_directory = (
            await session.scalars(
                sa.select(Directory).where(
                    sa.and_(
                        Directory.repository_id == repository.id,
                        Directory.name == part,
                        (
                            Directory.parent_id == current_directory.id
                            if current_directory
                            else Directory.parent_id.is_(None)
                        ),
                    )
                )
            )
        ).first()

        if not current_directory:
            return None

    return current_directory
copy async
copy(target, session, config, force=False, shallow=False)
PARAMETER DESCRIPTION
target

TYPE: Optional[Directory]

session

TYPE: SessionContext

config

TYPE: Config

force

TYPE: bool DEFAULT: False

shallow

TYPE: bool DEFAULT: False

Source code in src/materia/models/directory.py
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
async def copy(
    self,
    target: Optional["Directory"],
    session: SessionContext,
    config: Config,
    force: bool = False,
    shallow: bool = False,
) -> Self:
    session.add(self)
    await session.refresh(self, attribute_names=["repository"])

    repository_path = await self.repository.real_path(session, config)
    directory_path = await self.real_path(session, config)
    target_path = (
        await target.real_path(session, config) if target else repository_path
    )

    current_directory = FileSystem(directory_path, repository_path)
    new_directory = await current_directory.copy(
        target_path, force=force, shallow=shallow
    )

    cloned = self.clone()
    cloned.name = new_directory.name()
    cloned.parent_id = target.id if target else None
    session.add(cloned)
    await session.flush()

    await session.refresh(self, attribute_names=["files", "directories"])
    for directory in self.directories:
        await directory.copy(cloned, session, config, shallow=True)
    for file in self.files:
        await file.copy(cloned, session, config, shallow=True)

    return self
move async
move(target, session, config, force=False, shallow=False)
PARAMETER DESCRIPTION
target

TYPE: Optional[Directory]

session

TYPE: SessionContext

config

TYPE: Config

force

TYPE: bool DEFAULT: False

shallow

TYPE: bool DEFAULT: False

Source code in src/materia/models/directory.py
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
async def move(
    self,
    target: Optional["Directory"],
    session: SessionContext,
    config: Config,
    force: bool = False,
    shallow: bool = False,
) -> Self:
    session.add(self)
    await session.refresh(self, attribute_names=["repository"])

    repository_path = await self.repository.real_path(session, config)
    directory_path = await self.real_path(session, config)
    target_path = (
        await target.real_path(session, config) if target else repository_path
    )

    current_directory = FileSystem(directory_path, repository_path)
    moved_directory = await current_directory.move(
        target_path, force=force, shallow=shallow
    )

    self.name = moved_directory.name()
    self.parent_id = target.id if target else None
    self.updated = time()

    await session.flush()

    return self
rename async
rename(name, session, config, force=False, shallow=False)
PARAMETER DESCRIPTION
name

TYPE: str

session

TYPE: SessionContext

config

TYPE: Config

force

TYPE: bool DEFAULT: False

shallow

TYPE: bool DEFAULT: False

Source code in src/materia/models/directory.py
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
async def rename(
    self,
    name: str,
    session: SessionContext,
    config: Config,
    force: bool = False,
    shallow: bool = False,
) -> Self:
    session.add(self)
    await session.refresh(self, attribute_names=["repository"])

    repository_path = await self.repository.real_path(session, config)
    directory_path = await self.real_path(session, config)

    current_directory = FileSystem(directory_path, repository_path)
    renamed_directory = await current_directory.rename(
        name, force=force, shallow=shallow
    )

    self.name = renamed_directory.name()
    await session.flush()
    return self
info async
info(session)
PARAMETER DESCRIPTION
session

TYPE: SessionContext

Source code in src/materia/models/directory.py
245
246
247
248
249
250
251
252
253
254
255
256
async def info(self, session: SessionContext) -> "DirectoryInfo":
    session.add(self)
    await session.refresh(self, attribute_names=["files"])

    info = DirectoryInfo.model_validate(self)

    relative_path = await self.relative_path(session)

    info.path = Path("/").joinpath(relative_path) if relative_path else None
    info.used = sum([file.size for file in self.files])

    return info

Bases: Base

id class-attribute instance-attribute
id = mapped_column(BigInteger, primary_key=True)
directory_id class-attribute instance-attribute
directory_id = mapped_column(
    ForeignKey("directory.id", ondelete="CASCADE")
)
created class-attribute instance-attribute
created = mapped_column(BigInteger, default=time)
url instance-attribute
url
directory class-attribute instance-attribute
directory = relationship(back_populates='link')
to_dict
to_dict()
Source code in src/materia/models/base.py
7
8
def to_dict(self) -> dict:
    return {key: getattr(self, key) for key in self.__table__.columns.keys()}
clone
clone()

Clone model. Included: columns and values, foreign keys Ignored: primary keys, relationships

Source code in src/materia/models/base.py
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
def clone(self) -> Optional[Self]:
    """Clone model.
    Included: columns and values, foreign keys
    Ignored: primary keys, relationships
    """
    # if not inspect(self).persistent:
    #    return

    cloned = self.__class__(
        **{
            key: getattr(self, key)
            for key in self.__table__.columns.keys()
            # ignore primary keys
            if key not in self.__table__.primary_key.columns.keys()
        }
    )

    return cloned

DirectoryInfo

Bases: BaseModel

model_config class-attribute instance-attribute
model_config = ConfigDict(from_attributes=True)
id instance-attribute
id
repository_id instance-attribute
repository_id
parent_id instance-attribute
parent_id
created instance-attribute
created
updated instance-attribute
updated
name instance-attribute
name
is_public instance-attribute
is_public
path class-attribute instance-attribute
path = None
used class-attribute instance-attribute
used = None

DirectoryContent

Bases: BaseModel

model_config class-attribute instance-attribute
model_config = ConfigDict(arbitrary_types_allowed=True)
files instance-attribute
files
directories instance-attribute
directories

DirectoryPath

Bases: BaseModel

path instance-attribute
path

DirectoryRename

Bases: BaseModel

path instance-attribute
path
name instance-attribute
name
force class-attribute instance-attribute
force = False

DirectoryCopyMove

Bases: BaseModel

path instance-attribute
path
target instance-attribute
target
force class-attribute instance-attribute
force = False

file

FileError

Bases: Exception

File

Bases: Base

id class-attribute instance-attribute
id = mapped_column(BigInteger, primary_key=True)
repository_id class-attribute instance-attribute
repository_id = mapped_column(
    ForeignKey("repository.id", ondelete="CASCADE")
)
parent_id class-attribute instance-attribute
parent_id = mapped_column(
    ForeignKey("directory.id", ondelete="CASCADE"),
    nullable=True,
)
created class-attribute instance-attribute
created = mapped_column(
    BigInteger, nullable=False, default=time
)
updated class-attribute instance-attribute
updated = mapped_column(
    BigInteger, nullable=False, default=time
)
name instance-attribute
name
is_public class-attribute instance-attribute
is_public = mapped_column(default=False)
size class-attribute instance-attribute
size = mapped_column(BigInteger, nullable=True)
repository class-attribute instance-attribute
repository = relationship(back_populates='files')
parent class-attribute instance-attribute
parent = relationship(back_populates='files')
link = relationship(back_populates='file')
to_dict
to_dict()
Source code in src/materia/models/base.py
7
8
def to_dict(self) -> dict:
    return {key: getattr(self, key) for key in self.__table__.columns.keys()}
clone
clone()

Clone model. Included: columns and values, foreign keys Ignored: primary keys, relationships

Source code in src/materia/models/base.py
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
def clone(self) -> Optional[Self]:
    """Clone model.
    Included: columns and values, foreign keys
    Ignored: primary keys, relationships
    """
    # if not inspect(self).persistent:
    #    return

    cloned = self.__class__(
        **{
            key: getattr(self, key)
            for key in self.__table__.columns.keys()
            # ignore primary keys
            if key not in self.__table__.primary_key.columns.keys()
        }
    )

    return cloned
new async
new(data, session, config)
PARAMETER DESCRIPTION
data

TYPE: Union[bytes, Path]

session

TYPE: SessionContext

config

TYPE: Config

Source code in src/materia/models/file.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 new(
    self, data: Union[bytes, Path], session: SessionContext, config: Config
) -> Optional[Self]:
    session.add(self)
    await session.flush()
    await session.refresh(self, attribute_names=["repository"])

    file_path = await self.real_path(session, config)
    repository_path = await self.repository.real_path(session, config)
    new_file = FileSystem(file_path, repository_path)

    if isinstance(data, bytes):
        await new_file.write_file(data)
    elif isinstance(data, Path):
        from_file = FileSystem(data, config.application.working_directory)
        await from_file.move(file_path.parent, new_name=file_path.name)
    else:
        raise FileError(f"Unknown data type passed: {type(data)}")

    self.size = await new_file.size()
    await session.flush()

    return self
remove async
remove(session, config)
PARAMETER DESCRIPTION
session

TYPE: SessionContext

config

TYPE: Config

Source code in src/materia/models/file.py
62
63
64
65
66
67
68
69
70
71
72
73
async def remove(self, session: SessionContext, config: Config):
    session.add(self)

    file_path = await self.real_path(session, config)

    new_file = FileSystem(
        file_path, await self.repository.real_path(session, config)
    )
    await new_file.remove()

    await session.delete(self)
    await session.flush()
relative_path async
relative_path(session)
PARAMETER DESCRIPTION
session

TYPE: SessionContext

Source code in src/materia/models/file.py
75
76
77
78
79
80
81
82
83
84
85
86
87
88
async def relative_path(self, session: SessionContext) -> Optional[Path]:
    if inspect(self).was_deleted:
        return None

    file_path = Path()

    async with session.begin_nested():
        session.add(self)
        await session.refresh(self, attribute_names=["parent"])

        if self.parent:
            file_path = await self.parent.relative_path(session)

    return file_path.joinpath(self.name)
real_path async
real_path(session, config)
PARAMETER DESCRIPTION
session

TYPE: SessionContext

config

TYPE: Config

Source code in src/materia/models/file.py
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
async def real_path(
    self, session: SessionContext, config: Config
) -> Optional[Path]:
    if inspect(self).was_deleted:
        return None

    file_path = Path()

    async with session.begin_nested():
        session.add(self)
        await session.refresh(self, attribute_names=["repository", "parent"])

        if self.parent:
            file_path = await self.parent.real_path(session, config)
        else:
            file_path = await self.repository.real_path(session, config)

    return file_path.joinpath(self.name)
by_path async staticmethod
by_path(repository, path, session, config)
PARAMETER DESCRIPTION
repository

TYPE: Repository

path

TYPE: Path

session

TYPE: SessionContext

config

TYPE: Config

Source code in src/materia/models/file.py
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
@staticmethod
async def by_path(
    repository: "Repository", path: Path, session: SessionContext, config: Config
) -> Optional[Self]:
    if path == Path():
        raise FileError("Cannot find file by empty path")

    parent_directory = (
        None
        if path.parent == Path()
        else await Directory.by_path(repository, path.parent, session, config)
    )

    current_file = (
        await session.scalars(
            sa.select(File).where(
                sa.and_(
                    File.repository_id == repository.id,
                    File.name == path.name,
                    (
                        File.parent_id == parent_directory.id
                        if parent_directory
                        else File.parent_id.is_(None)
                    ),
                )
            )
        )
    ).first()

    return current_file
copy async
copy(
    directory, session, config, force=False, shallow=False
)
PARAMETER DESCRIPTION
directory

TYPE: Optional[Directory]

session

TYPE: SessionContext

config

TYPE: Config

force

TYPE: bool DEFAULT: False

shallow

TYPE: bool DEFAULT: False

Source code in src/materia/models/file.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
async def copy(
    self,
    directory: Optional["Directory"],
    session: SessionContext,
    config: Config,
    force: bool = False,
    shallow: bool = False,
) -> Self:
    session.add(self)
    await session.refresh(self, attribute_names=["repository"])

    repository_path = await self.repository.real_path(session, config)
    file_path = await self.real_path(session, config)
    directory_path = (
        await directory.real_path(session, config) if directory else repository_path
    )

    current_file = FileSystem(file_path, repository_path)
    new_file = await current_file.copy(directory_path, force=force, shallow=shallow)

    cloned = self.clone()
    cloned.name = new_file.name()
    cloned.parent_id = directory.id if directory else None
    session.add(cloned)
    await session.flush()

    return self
move async
move(
    directory, session, config, force=False, shallow=False
)
PARAMETER DESCRIPTION
directory

TYPE: Optional[Directory]

session

TYPE: SessionContext

config

TYPE: Config

force

TYPE: bool DEFAULT: False

shallow

TYPE: bool DEFAULT: False

Source code in src/materia/models/file.py
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
async def move(
    self,
    directory: Optional["Directory"],
    session: SessionContext,
    config: Config,
    force: bool = False,
    shallow: bool = False,
) -> Self:
    session.add(self)
    await session.refresh(self, attribute_names=["repository"])

    repository_path = await self.repository.real_path(session, config)
    file_path = await self.real_path(session, config)
    directory_path = (
        await directory.real_path(session, config) if directory else repository_path
    )

    current_file = FileSystem(file_path, repository_path)
    moved_file = await current_file.move(
        directory_path, force=force, shallow=shallow
    )

    self.name = moved_file.name()
    self.parent_id = directory.id if directory else None
    self.updated = time()
    await session.flush()

    return self
rename async
rename(name, session, config, force=False, shallow=False)
PARAMETER DESCRIPTION
name

TYPE: str

session

TYPE: SessionContext

config

TYPE: Config

force

TYPE: bool DEFAULT: False

shallow

TYPE: bool DEFAULT: False

Source code in src/materia/models/file.py
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
async def rename(
    self,
    name: str,
    session: SessionContext,
    config: Config,
    force: bool = False,
    shallow: bool = False,
) -> Self:
    session.add(self)
    await session.refresh(self, attribute_names=["repository"])

    repository_path = await self.repository.real_path(session, config)
    file_path = await self.real_path(session, config)

    current_file = FileSystem(file_path, repository_path)
    renamed_file = await current_file.rename(name, force=force, shallow=shallow)

    self.name = renamed_file.name()
    self.updated = time()
    await session.flush()
    return self
info async
info(session)
PARAMETER DESCRIPTION
session

TYPE: SessionContext

Source code in src/materia/models/file.py
219
220
221
222
223
224
async def info(self, session: SessionContext) -> Optional["FileInfo"]:
    info = FileInfo.model_validate(self)
    relative_path = await self.relative_path(session)
    info.path = Path("/").joinpath(relative_path) if relative_path else None

    return info

Bases: Base

id class-attribute instance-attribute
id = mapped_column(BigInteger, primary_key=True)
file_id class-attribute instance-attribute
file_id = mapped_column(
    ForeignKey("file.id", ondelete="CASCADE")
)
created class-attribute instance-attribute
created = mapped_column(BigInteger, default=time)
url instance-attribute
url
file class-attribute instance-attribute
file = relationship(back_populates='link')
to_dict
to_dict()
Source code in src/materia/models/base.py
7
8
def to_dict(self) -> dict:
    return {key: getattr(self, key) for key in self.__table__.columns.keys()}
clone
clone()

Clone model. Included: columns and values, foreign keys Ignored: primary keys, relationships

Source code in src/materia/models/base.py
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
def clone(self) -> Optional[Self]:
    """Clone model.
    Included: columns and values, foreign keys
    Ignored: primary keys, relationships
    """
    # if not inspect(self).persistent:
    #    return

    cloned = self.__class__(
        **{
            key: getattr(self, key)
            for key in self.__table__.columns.keys()
            # ignore primary keys
            if key not in self.__table__.primary_key.columns.keys()
        }
    )

    return cloned

FileInfo

Bases: BaseModel

model_config class-attribute instance-attribute
model_config = ConfigDict(from_attributes=True)
id instance-attribute
id
repository_id instance-attribute
repository_id
parent_id instance-attribute
parent_id
created instance-attribute
created
updated instance-attribute
updated
name instance-attribute
name
is_public instance-attribute
is_public
size instance-attribute
size
path class-attribute instance-attribute
path = None

FilePath

Bases: BaseModel

path instance-attribute
path

FileRename

Bases: BaseModel

path instance-attribute
path
name instance-attribute
name
force class-attribute instance-attribute
force = False

FileCopyMove

Bases: BaseModel

path instance-attribute
path
target instance-attribute
target
force class-attribute instance-attribute
force = False

convert_bytes

convert_bytes(size)
PARAMETER DESCRIPTION
size

TYPE: int

Source code in src/materia/models/file.py
227
228
229
230
231
def convert_bytes(size: int):
    for unit in ["bytes", "kB", "MB", "GB", "TB"]:
        if size < 1024:
            return f"{size}{unit}" if unit == "bytes" else f"{size:.1f}{unit}"
        size >>= 10

repository

RepositoryError

Bases: Exception

Repository

Bases: Base

id class-attribute instance-attribute
id = mapped_column(BigInteger, primary_key=True)
user_id class-attribute instance-attribute
user_id = mapped_column(
    ForeignKey("user.id", ondelete="CASCADE")
)
capacity class-attribute instance-attribute
capacity = mapped_column(BigInteger, nullable=False)
user class-attribute instance-attribute
user = relationship(back_populates='repository')
directories class-attribute instance-attribute
directories = relationship(back_populates='repository')
files class-attribute instance-attribute
files = relationship(back_populates='repository')
to_dict
to_dict()
Source code in src/materia/models/base.py
7
8
def to_dict(self) -> dict:
    return {key: getattr(self, key) for key in self.__table__.columns.keys()}
clone
clone()

Clone model. Included: columns and values, foreign keys Ignored: primary keys, relationships

Source code in src/materia/models/base.py
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
def clone(self) -> Optional[Self]:
    """Clone model.
    Included: columns and values, foreign keys
    Ignored: primary keys, relationships
    """
    # if not inspect(self).persistent:
    #    return

    cloned = self.__class__(
        **{
            key: getattr(self, key)
            for key in self.__table__.columns.keys()
            # ignore primary keys
            if key not in self.__table__.primary_key.columns.keys()
        }
    )

    return cloned
new async
new(session, config)
PARAMETER DESCRIPTION
session

TYPE: SessionContext

config

TYPE: Config

Source code in src/materia/models/repository.py
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
async def new(self, session: SessionContext, config: Config) -> Optional[Self]:
    session.add(self)
    await session.flush()

    repository_path = await self.real_path(session, config)
    relative_path = repository_path.relative_to(
        config.application.working_directory
    )

    try:
        repository_path.mkdir(parents=True, exist_ok=True)
    except OSError as e:
        raise RepositoryError(
            f"Failed to create repository at /{relative_path}:",
            *e.args,
        )

    await session.flush()

    return self
real_path async
real_path(session, config)

Get absolute path of the directory.

PARAMETER DESCRIPTION
session

TYPE: SessionContext

config

TYPE: Config

Source code in src/materia/models/repository.py
51
52
53
54
55
56
57
58
59
60
async def real_path(self, session: SessionContext, config: Config) -> Path:
    """Get absolute path of the directory."""
    session.add(self)
    await session.refresh(self, attribute_names=["user"])

    repository_path = config.application.working_directory.joinpath(
        "repository", self.user.lower_name
    )

    return repository_path
remove async
remove(session, config)
PARAMETER DESCRIPTION
session

TYPE: SessionContext

config

TYPE: Config

Source code in src/materia/models/repository.py
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
async def remove(self, session: SessionContext, config: Config):
    session.add(self)
    await session.refresh(self, attribute_names=["directories", "files"])

    for directory in self.directories:
        if directory.is_root():
            await directory.remove(session)

    for file in self.files:
        await file.remove(session)

    repository_path = await self.real_path(session, config)

    try:
        shutil.rmtree(str(repository_path))
    except OSError as e:
        raise RepositoryError(
            f"Failed to remove repository at /{repository_path.relative_to(config.application.working_directory)}:",
            *e.args,
        )

    await session.delete(self)
    await session.flush()
update async
update(session)
PARAMETER DESCRIPTION
session

TYPE: SessionContext

Source code in src/materia/models/repository.py
86
87
88
89
90
async def update(self, session: SessionContext):
    await session.execute(
        sa.update(Repository).values(self.to_dict()).where(Repository.id == self.id)
    )
    await session.flush()
from_user async staticmethod
from_user(user, session)
PARAMETER DESCRIPTION
user

TYPE: User

session

TYPE: SessionContext

Source code in src/materia/models/repository.py
92
93
94
95
96
@staticmethod
async def from_user(user: "User", session: SessionContext) -> Optional[Self]:
    session.add(user)
    await session.refresh(user, attribute_names=["repository"])
    return user.repository
used_capacity async
used_capacity(session)
PARAMETER DESCRIPTION
session

TYPE: SessionContext

Source code in src/materia/models/repository.py
 98
 99
100
101
102
async def used_capacity(self, session: SessionContext) -> int:
    session.add(self)
    await session.refresh(self, attribute_names=["files"])

    return sum([file.size for file in self.files])
remaining_capacity async
remaining_capacity(session)
PARAMETER DESCRIPTION
session

TYPE: SessionContext

Source code in src/materia/models/repository.py
104
105
106
async def remaining_capacity(self, session: SessionContext) -> int:
    used = await self.used_capacity(session)
    return self.capacity - used
info async
info(session)
PARAMETER DESCRIPTION
session

TYPE: SessionContext

Source code in src/materia/models/repository.py
108
109
110
111
112
async def info(self, session: SessionContext) -> "RepositoryInfo":
    info = RepositoryInfo.model_validate(self)
    info.used = await self.used_capacity(session)

    return info

RepositoryInfo

Bases: BaseModel

model_config class-attribute instance-attribute
model_config = ConfigDict(from_attributes=True)
id instance-attribute
id
capacity instance-attribute
capacity
used class-attribute instance-attribute
used = None

RepositoryContent

Bases: BaseModel

model_config class-attribute instance-attribute
model_config = ConfigDict(arbitrary_types_allowed=True)
files instance-attribute
files
directories instance-attribute
directories

user

valid_username module-attribute

valid_username = compile('^[\\da-zA-Z][-.\\w]*$')

invalid_username module-attribute

invalid_username = compile('[-._]{2,}|[-._]$')

UserError

Bases: Exception

User

Bases: Base

id class-attribute instance-attribute
id = mapped_column(primary_key=True, default=uuid4)
name class-attribute instance-attribute
name = mapped_column(unique=True)
lower_name class-attribute instance-attribute
lower_name = mapped_column(unique=True)
full_name instance-attribute
full_name
email instance-attribute
email
is_email_private class-attribute instance-attribute
is_email_private = mapped_column(default=True)
hashed_password instance-attribute
hashed_password
must_change_password class-attribute instance-attribute
must_change_password = mapped_column(default=False)
login_type instance-attribute
login_type
created class-attribute instance-attribute
created = mapped_column(BigInteger, default=time)
updated class-attribute instance-attribute
updated = mapped_column(BigInteger, default=time)
last_login class-attribute instance-attribute
last_login = mapped_column(BigInteger, nullable=True)
is_active class-attribute instance-attribute
is_active = mapped_column(default=False)
is_admin class-attribute instance-attribute
is_admin = mapped_column(default=False)
avatar instance-attribute
avatar
repository class-attribute instance-attribute
repository = relationship(back_populates='user')
to_dict
to_dict()
Source code in src/materia/models/base.py
7
8
def to_dict(self) -> dict:
    return {key: getattr(self, key) for key in self.__table__.columns.keys()}
clone
clone()

Clone model. Included: columns and values, foreign keys Ignored: primary keys, relationships

Source code in src/materia/models/base.py
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
def clone(self) -> Optional[Self]:
    """Clone model.
    Included: columns and values, foreign keys
    Ignored: primary keys, relationships
    """
    # if not inspect(self).persistent:
    #    return

    cloned = self.__class__(
        **{
            key: getattr(self, key)
            for key in self.__table__.columns.keys()
            # ignore primary keys
            if key not in self.__table__.primary_key.columns.keys()
        }
    )

    return cloned
new async
new(session, config)
PARAMETER DESCRIPTION
session

TYPE: SessionContext

config

TYPE: Config

Source code in src/materia/models/user.py
52
53
54
55
56
57
async def new(self, session: SessionContext, config: Config) -> Optional[Self]:
    # Provide checks outer

    session.add(self)
    await session.flush()
    return self
remove async
remove(session)
PARAMETER DESCRIPTION
session

TYPE: SessionContext

Source code in src/materia/models/user.py
59
60
61
62
63
64
65
66
67
async def remove(self, session: SessionContext):
    session.add(self)
    await session.refresh(self, attribute_names=["repository"])

    if self.repository:
        await self.repository.remove()

    await session.delete(self)
    await session.flush()
update_last_login
update_last_login()
Source code in src/materia/models/user.py
69
70
def update_last_login(self):
    self.last_login = int(time.time())
is_local
is_local()
Source code in src/materia/models/user.py
72
73
def is_local(self) -> bool:
    return self.login_type == LoginType.Plain
is_oauth2
is_oauth2()
Source code in src/materia/models/user.py
75
76
def is_oauth2(self) -> bool:
    return self.login_type == LoginType.OAuth2
check_username staticmethod
check_username(name)
PARAMETER DESCRIPTION
name

TYPE: str

Source code in src/materia/models/user.py
78
79
80
@staticmethod
def check_username(name: str) -> bool:
    return bool(valid_username.match(name) and not invalid_username.match(name))
check_password staticmethod
check_password(password, config)
PARAMETER DESCRIPTION
password

TYPE: str

config

TYPE: Config

Source code in src/materia/models/user.py
82
83
84
@staticmethod
def check_password(password: str, config: Config) -> bool:
    return not len(password) < config.security.password_min_length
count async staticmethod
count(session)
PARAMETER DESCRIPTION
session

TYPE: SessionContext

Source code in src/materia/models/user.py
86
87
88
@staticmethod
async def count(session: SessionContext) -> Optional[int]:
    return await session.scalar(sa.select(sa.func.count(User.id)))
by_name async staticmethod
by_name(name, session, with_lower=False)
PARAMETER DESCRIPTION
name

TYPE: str

session

TYPE: SessionContext

with_lower

TYPE: bool DEFAULT: False

Source code in src/materia/models/user.py
90
91
92
93
94
95
96
97
98
@staticmethod
async def by_name(
    name: str, session: SessionContext, with_lower: bool = False
) -> Optional[Self]:
    if with_lower:
        query = User.lower_name == name.lower()
    else:
        query = User.name == name
    return (await session.scalars(sa.select(User).where(query))).first()
by_email async staticmethod
by_email(email, session)
PARAMETER DESCRIPTION
email

TYPE: str

session

TYPE: SessionContext

Source code in src/materia/models/user.py
100
101
102
103
104
@staticmethod
async def by_email(email: str, session: SessionContext) -> Optional[Self]:
    return (
        await session.scalars(sa.select(User).where(User.email == email))
    ).first()
by_id async staticmethod
by_id(id, session)
PARAMETER DESCRIPTION
id

TYPE: UUID

session

TYPE: SessionContext

Source code in src/materia/models/user.py
106
107
108
@staticmethod
async def by_id(id: UUID, session: SessionContext) -> Optional[Self]:
    return (await session.scalars(sa.select(User).where(User.id == id))).first()
edit_name async
edit_name(name, session)
PARAMETER DESCRIPTION
name

TYPE: str

session

TYPE: SessionContext

Source code in src/materia/models/user.py
110
111
112
113
114
115
116
117
118
119
async def edit_name(self, name: str, session: SessionContext) -> Self:
    if not User.check_username(name):
        raise UserError(f"Invalid username: {name}")

    self.name = name
    self.lower_name = name.lower()
    session.add(self)
    await session.flush()

    return self
edit_password async
edit_password(password, session, config)
PARAMETER DESCRIPTION
password

TYPE: str

session

TYPE: SessionContext

config

TYPE: Config

Source code in src/materia/models/user.py
121
122
123
124
125
126
127
128
129
130
131
132
133
134
async def edit_password(
    self, password: str, session: SessionContext, config: Config
) -> Self:
    if not User.check_password(password, config):
        raise UserError("Invalid password")

    self.hashed_password = security.hash_password(
        password, algo=config.security.password_hash_algo
    )

    session.add(self)
    await session.flush()

    return self
edit_email async
edit_email()
Source code in src/materia/models/user.py
136
137
async def edit_email(self):
    pass
info
info()
Source code in src/materia/models/user.py
139
140
141
142
def info(self) -> "UserInfo":
    user_info = UserInfo.model_validate(self)

    return user_info
edit_avatar async
edit_avatar(avatar, session, config)
PARAMETER DESCRIPTION
avatar

TYPE: BinaryIO | None

session

TYPE: SessionContext

config

TYPE: Config

Source code in src/materia/models/user.py
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
async def edit_avatar(
    self, avatar: BinaryIO | None, session: SessionContext, config: Config
):
    avatar_dir = config.application.working_directory.joinpath("avatars")

    if avatar is None:
        if self.avatar is None:
            return

        avatar_file = FileSystem(
            avatar_dir.joinpath(self.avatar), config.application.working_directory
        )
        if await avatar_file.exists():
            await avatar_file.remove()

        session.add(self)
        self.avatar = None
        await session.flush()

        return

    try:
        image = Image.open(avatar)
    except Exception as e:
        raise UserError("Failed to read avatar data") from e

    avatar_hashes: list[str] = (
        await session.scalars(sa.select(User.avatar).where(User.avatar.isnot(None)))
    ).all()
    avatar_id = Sqids(min_length=10, blocklist=avatar_hashes).encode(
        [int(time.time())]
    )

    try:
        if not avatar_dir.exists():
            await async_os.mkdir(avatar_dir)
        image.save(avatar_dir.joinpath(avatar_id), format=image.format)
    except Exception as e:
        raise UserError(f"Failed to save avatar: {e}") from e

    if old_avatar := self.avatar:
        avatar_file = FileSystem(
            avatar_dir.joinpath(old_avatar), config.application.working_directory
        )
        if await avatar_file.exists():
            await avatar_file.remove()

    session.add(self)
    self.avatar = avatar_id
    await session.flush()

UserCredentials

Bases: BaseModel

name instance-attribute
name
password instance-attribute
password
email instance-attribute
email

UserInfo

Bases: BaseModel

model_config class-attribute instance-attribute
model_config = ConfigDict(from_attributes=True)
id instance-attribute
id
name instance-attribute
name
lower_name instance-attribute
lower_name
full_name instance-attribute
full_name
email instance-attribute
email
is_email_private instance-attribute
is_email_private
must_change_password instance-attribute
must_change_password
login_type instance-attribute
login_type
created instance-attribute
created
updated instance-attribute
updated
last_login instance-attribute
last_login
is_active instance-attribute
is_active
is_admin instance-attribute
is_admin
avatar instance-attribute
avatar