file.collection

Attributes

FileT

FileSetT

Classes

FileCollection

Manages files.

FileSetCollection

Manages filesets.

Module Contents

file.collection.FileT[source]
file.collection.FileSetT[source]
class file.collection.FileCollection(session: sqlalchemy.orm.Session, type: Literal['*', 'generic'] = '*', allow_duplicates: bool = True)[source]
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.

query() Query[FileT][source]
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[source]

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[source]

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

assert_not_duplicate(content: bytes | IO[bytes]) None[source]
delete(file: onegov.file.models.File) None[source]
no_longer_published_files(horizon: datetime.datetime) Query[FileT][source]

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

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

Returns a query of files which may be published.

publish_files(horizon: datetime | None = None) None[source]

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

by_id(file_id: str) FileT | None[source]

Returns the file with the given id or None.

by_filename(filename: str) Query[FileT][source]

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][source]

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

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

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

by_content_type(content_type: str) Query[FileT][source]

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

by_signature_digest(digest: str) Query[FileT][source]

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[source]

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'] = '*')[source]
class file.collection.FileSetCollection(session: sqlalchemy.orm.Session, type: str)

Bases: Generic[FileSetT]

Manages filesets.

query() Query[FileSetT][source]
add(title: str, meta: dict[str, Any] | None = None, content: dict[str, Any] | None = None) FileSetT[source]
delete(fileset: onegov.file.models.FileSet) None[source]
by_id(fileset_id: str) FileSetT | None[source]

Returns the fileset with the given id or None.