file.collection

Attributes

FileT
FileSetT

Classes

FileCollection	Manages files.
FileSetCollection	Manages filesets.

Module Contents

file.collection.FileT

file.collection.FileSetT

class file.collection.FileCollection(session: sqlalchemy.orm.Session, type: Literal['*', 'generic'] = '*', allow_duplicates: bool = True)

class file.collection.FileCollection(session: sqlalchemy.orm.Session, type: str, allow_duplicates: bool = True)

Bases: Generic[FileT]

Manages files.

Parameters:

• session – The SQLAlchemy db session to use.

• type – The polymorphic type to use and to filter for, or ‘*’ for all.

• allow_duplicates –

  Prevents duplicates if set to false. Duplicates are detected before pre-processing, so already stored files may be downloaded and added again, as they might have changed during the upload.

  Note that this does not change existing files. It only prevents new duplicates from being added.

session

type = '*'

allow_duplicates = True

query() → Query[FileT]

add(filename: str, content: bytes | IO[bytes], note: str | None = None, published: bool = True, publish_date: datetime | None = None, publish_end_date: datetime | None = None) → FileT

Adds a file with the given filename. The content maybe either in bytes or a file object.

replace(file: onegov.file.models.File, content: bytes | IO[bytes]) → None

Replaces the content of the given file with the new content.

assert_not_duplicate(content: bytes | IO[bytes]) → None

delete(file: onegov.file.models.File) → None

no_longer_published_files(horizon: datetime.datetime) → Query[FileT]

Returns a query of files where the publishing end date has expired.

publishable_files(horizon: datetime.datetime) → Query[FileT]

Returns a query of files which may be published.

publish_files(horizon: datetime | None = None) → None

Publishes unpublished files with a publish date older than the given horizon.

by_id(file_id: str) → FileT | None

Returns the file with the given id or None.

by_filename(filename: str) → Query[FileT]

Returns a query that matches the files with the given filename.

Be aware that there may be multiple files with the same filename!

by_checksum(checksum: str) → Query[FileT]

Returns a query that matches the given checksum (may be more than one record).

by_content(content: bytes | IO[bytes]) → Query[FileT]

Returns a query that matches the given content (may be more than one record).

by_content_type(content_type: str) → Query[FileT]

Returns a query that matches the given MIME content type (may be more than one record).

by_signature_digest(digest: str) → Query[FileT]

Returns a query that matches the given digest in the signature metadata. In other words, given a digest this function will find signed files that match the digest - either before or after signing.

Unsigned files are ignored.

The digest is expected to be a SHA256 hex.

locate_signature_metadata(digest: str) → SignatureMetadata | None

Looks for the given digest in the files table - if that doesn’t work it will go through the audit trail (i.e. the chat messages) and see if the digest can be found there.

If this database was ever used to sign a file with the given digest, or if a file that was signed had the given digest, this function will find it - barring manual database manipulation in the messages log.

class file.collection.FileSetCollection(session: sqlalchemy.orm.Session, type: Literal['*', 'generic'] = '*')

class file.collection.FileSetCollection(session: sqlalchemy.orm.Session, type: str)

Bases: Generic[FileSetT]

Manages filesets.

session

type = '*'

query() → Query[FileSetT]

add(title: str, meta: dict[str, Any] | None = None, content: dict[str, Any] | None = None) → FileSetT

delete(fileset: onegov.file.models.FileSet) → None

by_id(fileset_id: str) → FileSetT | None

Returns the fileset with the given id or None.