126 lines
4.4 KiB
Python
126 lines
4.4 KiB
Python
from __future__ import annotations
|
|
|
|
from aiobotocore.session import get_session
|
|
|
|
|
|
class S3Service:
|
|
def __init__(
|
|
self,
|
|
*,
|
|
bucket: str,
|
|
region: str,
|
|
access_key_id: str | None,
|
|
secret_access_key: str | None,
|
|
public_base_url: str | None,
|
|
endpoint_url: str | None,
|
|
use_reg_ru_website_public_host: bool,
|
|
):
|
|
self._bucket = bucket
|
|
self._region = region or 'us-east-1'
|
|
self._access_key_id = access_key_id
|
|
self._secret_access_key = secret_access_key
|
|
pb = (public_base_url or '').strip().rstrip('/')
|
|
self._public_base_url = pb if pb else None
|
|
self._endpoint_url = endpoint_url.strip().rstrip('/') if endpoint_url and endpoint_url.strip() else None
|
|
self._use_reg_ru_website_public_host = use_reg_ru_website_public_host
|
|
|
|
@staticmethod
|
|
def _url_prefix_variants(prefix: str) -> list[str]:
|
|
p = prefix.rstrip('/') + '/'
|
|
out = [p]
|
|
if p.startswith('https://'):
|
|
out.append('http://' + p[8:])
|
|
elif p.startswith('http://'):
|
|
out.append('https://' + p[7:])
|
|
return out
|
|
|
|
def _public_url_prefixes(self) -> list[str]:
|
|
acc: list[str] = []
|
|
pb = self._public_base_url
|
|
if pb:
|
|
acc.extend(self._url_prefix_variants(pb))
|
|
ep = self._endpoint_url
|
|
if ep:
|
|
base = f'{ep.rstrip("/")}/{self._bucket}'
|
|
acc.extend(self._url_prefix_variants(base))
|
|
if ep and self._use_reg_ru_website_public_host and 's3.regru.cloud' in ep.lower():
|
|
wh = f'https://{self._bucket}.website.regru.cloud'
|
|
acc.extend(self._url_prefix_variants(wh))
|
|
if not ep:
|
|
if self._region == 'us-east-1':
|
|
h = f'https://{self._bucket}.s3.amazonaws.com'
|
|
else:
|
|
h = f'https://{self._bucket}.s3.{self._region}.amazonaws.com'
|
|
acc.extend(self._url_prefix_variants(h))
|
|
seen: set[str] = set()
|
|
uniq: list[str] = []
|
|
for x in sorted(acc, key=len, reverse=True):
|
|
if x not in seen:
|
|
seen.add(x)
|
|
uniq.append(x)
|
|
return uniq
|
|
|
|
def object_key_from_public_url(self, url: str) -> str | None:
|
|
u = (url or '').strip()
|
|
if not u:
|
|
return None
|
|
for p in self._public_url_prefixes():
|
|
if u.startswith(p):
|
|
k = u[len(p):].split('?', 1)[0].split('#', 1)[0]
|
|
return k if k else None
|
|
return None
|
|
|
|
def _object_url(self, key: str) -> str:
|
|
if self._public_base_url:
|
|
return f'{self._public_base_url}/{key}'
|
|
endpoint = self._endpoint_url
|
|
if endpoint:
|
|
if (
|
|
self._use_reg_ru_website_public_host
|
|
and 's3.regru.cloud' in endpoint.lower()
|
|
):
|
|
return f'https://{self._bucket}.website.regru.cloud/{key}'
|
|
return f'{endpoint}/{self._bucket}/{key}'
|
|
region = self._region
|
|
if region == 'us-east-1':
|
|
host = 's3.amazonaws.com'
|
|
else:
|
|
host = f's3.{region}.amazonaws.com'
|
|
return f'https://{self._bucket}.{host}/{key}'
|
|
|
|
async def upload_bytes(self, *, key: str, body: bytes, content_type: str) -> str:
|
|
session = get_session()
|
|
kw: dict[str, object] = {'region_name': self._region}
|
|
aid = self._access_key_id
|
|
sk = self._secret_access_key
|
|
ep = self._endpoint_url
|
|
if aid:
|
|
kw['aws_access_key_id'] = aid
|
|
if sk:
|
|
kw['aws_secret_access_key'] = sk
|
|
if ep:
|
|
kw['endpoint_url'] = ep
|
|
async with session.create_client('s3', **kw) as client:
|
|
await client.put_object(
|
|
Bucket=self._bucket,
|
|
Key=key,
|
|
Body=body,
|
|
ContentType=content_type,
|
|
)
|
|
return self._object_url(key)
|
|
|
|
async def delete_object(self, *, key: str) -> None:
|
|
session = get_session()
|
|
kw: dict[str, object] = {'region_name': self._region}
|
|
aid = self._access_key_id
|
|
sk = self._secret_access_key
|
|
ep = self._endpoint_url
|
|
if aid:
|
|
kw['aws_access_key_id'] = aid
|
|
if sk:
|
|
kw['aws_secret_access_key'] = sk
|
|
if ep:
|
|
kw['endpoint_url'] = ep
|
|
async with session.create_client('s3', **kw) as client:
|
|
await client.delete_object(Bucket=self._bucket, Key=key)
|