granicus_archiver.aws

granicus_archiver.aws.client

class granicus_archiver.aws.client.SchedulersTD[source]

Bases: TypedDict

class granicus_archiver.aws.client.S3ClientKwargs[source]

Bases: TypedDict

class granicus_archiver.aws.client.ClientBase(config: Config)[source]

Bases: object

Base class for AWS clients

Parameters:

config (Config)

s3_resource: S3ServiceResource

The S3 service resource

s3_client: S3Client

The S3 client

bucket: Bucket

The S3 bucket

session: Session

The boto3 session

url_for_key(key: str | Path, scheme: str | None = None) → URL[source]

Get a URL for an S3 key within bucket

Parameters:

  • key (str | Path)

  • scheme (str | None)

Return type:

URL

async get_object(key: str | Path) → Object[source]

Get an S3 object within bucket by key

Parameters:

key (str | Path)

Return type:

Object

async iter_objects(prefix: Path | str) → AsyncGenerator[ObjectSummary, None][source]

Iterate over objects in bucket with a given (optional) prefix

Parameters:

prefix (Path | str)

Return type:

AsyncGenerator[ObjectSummary, None]

async object_exists(key: str | Path) → bool[source]

Check if an object exists in bucket by key

Parameters:

key (str | Path)

Return type:

bool

async get_object_tags(obj: Object | str | Path) → dict[str, str][source]

Get the tags for an S3 object

Parameters:

obj (Object | str | Path)

Return type:

dict[str, str]

async get_object_sha1(obj: Object | str | Path) → SHA1Hash | None[source]

Get the SHA1 hash for an S3 object

Parameters:

obj (Object | str | Path)

Return type:

SHA1Hash | None

async check_object_hash(obj: Object | str | Path, local_hash: SHA1Hash) → None[source]

Check the SHA1 hash of an S3 object against a local hash

Raises:

HashMismatchError – If the hashes do not match

Parameters:

  • obj (Object | str | Path)

  • local_hash (SHA1Hash)

Return type:

None

async upload(key: str | Path, local_filename: Path, wait_exists: bool = True, check_exists: bool = True, local_meta: FileMeta | None = None, acl: Literal['private', 'public-read', 'public-read-write', 'authenticated-read', 'aws-exec-read', 'bucket-owner-read', 'bucket-owner-full-control'] | None = None) → Object[source]
async upload(key: str | Path, local_filename: Path, wait_exists: bool = True, check_exists: bool = True, local_meta: FileMeta | None = None, acl: Literal['private', 'public-read', 'public-read-write', 'authenticated-read', 'aws-exec-read', 'bucket-owner-read', 'bucket-owner-full-control'] | None = None) → Object
async upload(key: str | Path, local_filename: Path, wait_exists: bool = False, check_exists: bool = True, local_meta: FileMeta | None = None, acl: Literal['private', 'public-read', 'public-read-write', 'authenticated-read', 'aws-exec-read', 'bucket-owner-read', 'bucket-owner-full-control'] | None = None) → str | Path

Upload a file to S3

Parameters:

  • key – The S3 key to upload the file to

  • local_filename – The local file to upload

  • wait_exists – If True, wait for the object to exist before returning

  • check_exists – If True, check if the object exists before uploading

  • local_meta – The metadata for the local file

  • acl – The ACL for the uploaded object

async download_object(key: str | Path, local_filename: Path) → None[source]

Download an S3 object to a local file

Parameters:

  • key (str | Path)

  • local_filename (Path)

Return type:

None

class granicus_archiver.aws.client.ClipClient(config: Config, max_clips: int = 8, scheduler_limit: int = 8)[source]

Bases: ClientBase

AWS Client to upload items from a model.ClipCollection

Parameters:
property upload_dir: Path

Root folder name (key prefix) for uploaded clips (alias for config.AWSConfig.clips_prefix)

get_clip_file_upload_path(clip: Clip, key: Literal['agenda', 'minutes', 'audio', 'video', 'chapters', 'agenda_packet']) → Path[source]

Get the uploaded filename for a clip asset (relative to upload_dir)

Parameters:

  • clip (Clip)

  • key (Literal['agenda', 'minutes', 'audio', 'video', 'chapters', 'agenda_packet'])

Return type:

Path

async upload_data_file() → None[source]

Upload the clips data file to S3

Return type:

None

async upload_clip(clip: Clip, *file_keys: Literal['agenda', 'minutes', 'audio', 'video', 'chapters', 'agenda_packet']) → bool[source]

Upload all assets for the given clip

Parameters:

  • clip (Clip)

  • file_keys (Literal['agenda', 'minutes', 'audio', 'video', 'chapters', 'agenda_packet'])

Return type:

bool

async check_clip_needs_upload(clip: Clip) → tuple[bool, Clip, set[Literal['agenda', 'minutes', 'audio', 'video', 'chapters', 'agenda_packet']]][source]

Check if the given clip hash any assets that need to be uploaded

Parameters:

clip (Clip)

Return type:

tuple[bool, Clip, set[Literal[‘agenda’, ‘minutes’, ‘audio’, ‘video’, ‘chapters’, ‘agenda_packet’]]]

async handle_upload_check_jobs() → None[source]

Wait for jobs from check_clip_needs_upload() and spawns their upload_clip() jobs

Return type:

None

async upload_all(clips: ClipCollection) → None[source]

Upload assets for the given clips (up to by max_clips)

Parameters:

clips (ClipCollection)

Return type:

None

class granicus_archiver.aws.client.LegistarClientBase(config: Config, max_clips: int, legistar_data: _ModelT | None = None)[source]

Bases: ClientBase, Generic[_GuidT, _ItemT, _ModelT], ABC

Parameters:

  • config (Config)

  • max_clips (int)

  • legistar_data (_ModelT)

max_clips: int

The maximum number of items to upload

legistar_data: _ModelT

A AbstractLegistarModel instance

get_file_upload_path(guid: _GuidT, uid: LegistarFileUID) → Path[source]

Get the uploaded filename for a legistar asset (relative to upload_dir)

Parameters:
Return type:

Path

async upload_data_file() → None[source]

Upload the legistar data file to S3

Return type:

None

async upload_legistar_file(guid: _GuidT, uid: LegistarFileUID, local_file: Path, local_meta: FileMeta) → bool[source]

Upload a single legistar file

Parameters:
Return type:

bool

async upload_legistar_item(guid: _GuidT, *uids: LegistarFileUID) → bool[source]

Upload all assets for the legistar item matching the given guid

Parameters:
Return type:

bool

async check_item_needs_upload(guid: _GuidT) → tuple[bool, _GuidT, set[LegistarFileUID]][source]

Check if the item matching guid has any local files that need to be uploaded

Parameters:

guid (_GuidT)

Return type:

tuple[bool, _GuidT, set[LegistarFileUID]]

async handle_upload_check_jobs() → None[source]

Wait for jobs from check_item_needs_upload() and spawns their upload_legistar_item() jobs

Return type:

None

async upload_all() → None[source]

Upload files for all items in legistar_data (up to by max_clips)

Return type:

None

class granicus_archiver.aws.client.LegistarClient(config: Config, max_clips: int, legistar_data: _ModelT | None = None)[source]

Bases: LegistarClientBase[GUID, DetailPageResult, LegistarData]

Client to upload items from LegistarData

Parameters:

  • config (Config)

  • max_clips (int)

  • legistar_data (_ModelT)

class granicus_archiver.aws.client.RGuidLegistarClient(config: Config, max_clips: int, legistar_data: _ModelT | None = None)[source]

Bases: LegistarClientBase[REAL_GUID, RGuidDetailResult, RGuidLegistarData]

Client to upload items from RGuidLegistarData

Parameters:

  • config (Config)

  • max_clips (int)

  • legistar_data (_ModelT)