Page MenuHomeSoftware Heritage
Differential D4843 Diff 17177 swh/lister/bitbucket/lister.py

Changeset View

Standalone View

View Options

swh/lister/bitbucket/lister.py

# Copyright (C) 2017-2019 The Software Heritage developers # Copyright (C) 2017-2019 The Software Heritage developers
# See the AUTHORS file at the top-level directory of this distribution # See the AUTHORS file at the top-level directory of this distribution
# License: GNU General Public License version 3, or any later version # License: GNU General Public License version 3, or any later version
# See top-level LICENSE file for more information # See top-level LICENSE file for more information
from datetime import datetime, timezone from dataclasses import asdict, dataclass
from datetime import datetime
import logging import logging
from typing import Any, Dict, List, Optional import time
from typing import Any, Dict, Iterator, List, Optional
from urllib import parse from urllib import parse
import iso8601 import iso8601
from requests import Response import requests
from swh.lister.bitbucket.models import BitBucketModel from swh.scheduler.interface import SchedulerInterface
ardumontUnsubmitted
Not Done
Inline Actions

Do we entirely drop the old bitbucket lister code?

If yes, you can also remove that module.

ardumont: Do we entirely drop the old bitbucket lister code? If yes, you can also remove that module.
from swh.lister.core.indexing_lister import IndexingHttpLister from swh.scheduler.model import ListedOrigin
from .. import USER_AGENT
from ..pattern import CredentialsType, Lister
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
class BitBucketLister(IndexingHttpLister): @dataclass
PATH_TEMPLATE = "/repositories?after=%s" class BitbucketListerState:
MODEL = BitBucketModel """State of my lister"""
last_repo_cdate: Optional[datetime] = None
"""Date and time of the last repository listed on an incremental pass"""
class BitbucketLister(Lister[BitbucketListerState, List[Dict[str, Any]]]):
"""List origins from Bitbucket.
"""
LISTER_NAME = "bitbucket" LISTER_NAME = "bitbucket"
DEFAULT_URL = "https://api.bitbucket.org/2.0" INSTANCE = "bitbucket"
instance = "bitbucket"
default_min_bound = datetime.fromtimestamp(0, timezone.utc) # type: Any API_URL = "https://api.bitbucket.org/2.0/repositories"
PAGE_SIZE = 100
BACKOFF_FACTOR = 10 # max anonymous 60 per hour, authenticated 1000 per hour
MAX_RETRIES = 5
def __init__( def __init__(
self, url: str = None, override_config=None, per_page: int = 100 self,
) -> None: scheduler: SchedulerInterface,
super().__init__(url=url, override_config=override_config) per_page: int = 100,
per_page = self.config.get("per_page", per_page) credentials: CredentialsType = None,
):
self.PATH_TEMPLATE = "%s&pagelen=%s" % (self.PATH_TEMPLATE, per_page) super().__init__(
scheduler=scheduler,
def get_model_from_repo(self, repo: Dict) -> Dict[str, Any]: credentials=credentials,
return { url=self.API_URL,
"uid": repo["uuid"], instance=self.INSTANCE,
"indexable": iso8601.parse_date(repo["created_on"]), )
"name": repo["name"],
"full_name": repo["full_name"], self.url_params = {
"html_url": repo["links"]["html"]["href"], "pagelen": per_page,
"origin_url": repo["links"]["clone"][0]["href"], "fields": (
"origin_type": repo["scm"], "next,values.links.clone.href,values.scm,values.updated_on,"
"values.created_on"
),
} }
def get_next_target_from_response(self, response: Response) -> Optional[datetime]: self.session = requests.Session()
"""This will read the 'next' link from the api response if any self.session.headers.update(
and return it as a datetime. {"Accept": "application/json", "User-Agent": USER_AGENT}
)
self.backoff = self.BACKOFF_FACTOR
self.request_count = 0
def state_from_dict(self, d: Dict[str, Any]) -> BitbucketListerState:
vlorentzUnsubmitted
Done
Inline Actions

maybe a warning if there is more than one set of credentials?

vlorentz: maybe a warning if there is more than one set of credentials?
last_repo_cdate = d.get("last_repo_cdate")
if last_repo_cdate is not None:
d["last_repo_cdate"] = iso8601.parse_date(last_repo_cdate)
return BitbucketListerState(**d)
def state_to_dict(self, state: BitbucketListerState) -> Dict[str, Any]:
d = asdict(state)
last_repo_cdate = d.get("last_repo_cdate")
if last_repo_cdate is not None:
d["last_repo_cdate"] = last_repo_cdate.isoformat()
return d
def set_session_credentials(self):
"""Updates session authentication headers with given credentials
vlorentzUnsubmitted
Done
Inline Actions

Shouldn't these be (class)methods of BitbucketListerState?

vlorentz: Shouldn't these be (class)methods of `BitbucketListerState`?
tenmaAuthorUnsubmitted
Done
Inline Actions

the new lister API has them as instance methods of the lister, but indeed the API could be amended to have it as ListerState class methods.
I would say, let discuss it and do that later.

tenma: the new lister API has them as instance methods of the lister, but indeed the API could be…
vlorentzUnsubmitted
Done
Inline Actions

yeah that makes sense, let's leave it like this then

vlorentz: yeah that makes sense, let's leave it like this then
Support only one credential of type basic auth"""
vlorentzUnsubmitted
Done
Inline Actions
return d
- def set_page_size(self, page_size: Optional[int]):
+ def set_page_size(self, page_size: Optional[int]) -> None:
"""Set number of bitbuckets origins to return per page."""
vlorentz:
if len(self.credentials) > 0:
anlambertUnsubmitted
Done
Inline Actions

Can you rename the function to set_credentials, shorter is better. Also parameters should be typed to Optional[str].

anlambert: Can you rename the function to `set_credentials`, shorter is better. Also parameters should be…
cred = self.credentials[0]
if "username" in cred:
self.session.auth = (cred["username"], cred["password"])
vlorentzUnsubmitted
Done
Inline Actions
self.url_params["pagelen"] = page_size
- def set_credentials(self, username: Optional[str], password: Optional[str]):
+ def set_credentials(self, username: Optional[str], password: Optional[str]) -> None:
"""Set basic authentication headers with given credentials."""
vlorentz:
def get_pages(self) -> Iterator[List[Dict[str, Any]]]:
anlambertUnsubmitted
Done
Inline Actions

you can remove that parameter as it is already the default value of the decorator

anlambert: you can remove that parameter as it is already the default value of the decorator
self.set_session_credentials()
vlorentzUnsubmitted
Done
Inline Actions
self.session.auth = (username, password)
- def set_incremental(self, incremental: Optional[bool]):
+ def set_incremental(self, incremental: Optional[bool]) -> None:
if incremental is not None:

And why does it need to be Optional[bool] instead of just bool?

vlorentz: And why does it need to be `Optional[bool]` instead of just `bool`?
last_repo_cdate: Optional[str] = "1970-01-01"
if self.state is not None and self.state.last_repo_cdate is not None:
last_repo_cdate = self.state.last_repo_cdate.isoformat()
while last_repo_cdate is not None:
self.url_params["after"] = last_repo_cdate
logger.debug("Fetching URL %s with params %s", self.url, self.url_params)
response = self.session.get(self.url, params=self.url_params)
# handle rate-limiting
if response.status_code == 429:
if self.request_count >= self.MAX_RETRIES:
logger.info(
"Max number of attempts hit (%s), giving up",
self.request_count,
)
break
logger.info("Rate limit was hit, sleeping %ss", self.backoff)
time.sleep(self.backoff)
self.backoff *= self.BACKOFF_FACTOR
self.request_count += 1
continue
# log other HTTP errors
if response.status_code != 200:
logger.warning(
"Got unexpected status_code %s: %s",
response.status_code,
response.content,
)
break
# response is OK, reset rate-limiting state
self.request_count = 0
self.backoff = self.BACKOFF_FACTOR
body = response.json()
next_page_url = body.get("next")
if next_page_url is not None:
next_page_url = parse.urlparse(next_page_url)
if not next_page_url.query:
logger.warning("Failed to parse url %s", next_page_url)
break
last_repo_cdate = parse.parse_qs(next_page_url.query)["after"][0]
else:
last_repo_cdate = None
Args: yield body["values"]
response (Response): requests' response from api call
Returns: def get_origins_from_page(
next date as a datetime self, page: List[Dict[str, Any]]
) -> Iterator[ListedOrigin]:
"""Convert a page of Bitbucket repositories into a list of ListedOrigins.
""" """
body = response.json() assert self.lister_obj.id is not None
next_ = body.get("next")
vlorentzUnsubmitted
Done
Inline Actions

IMO this yield should be moved before if next_page_url is not None for two reasons:

  1. we probably want to yield body["values"] even if we fail to parse next_page_url
  2. this means we can make last_repo_cdate non-optional and replace last_repo_cdate = None with a break (and make the loop a while True)
vlorentz: IMO this `yield` should be moved before `if next_page_url is not None` for two reasons: 1. we…
if next_ is not None: for repo in page:
next_ = parse.urlparse(next_) last_update = iso8601.parse_date(repo["updated_on"])
return iso8601.parse_date(parse.parse_qs(next_.query)["after"][0]) origin_url = repo["links"]["clone"][0]["href"]
return None origin_type = repo["scm"]
def transport_response_simplified(self, response: Response) -> List[Dict[str, Any]]: yield ListedOrigin(
repos = response.json()["values"] lister_id=self.lister_obj.id,
return [self.get_model_from_repo(repo) for repo in repos] url=origin_url,
visit_type=origin_type,
def request_uri(self, identifier: datetime) -> str: # type: ignore last_update=last_update,
identifier_str = parse.quote(identifier.isoformat()) )
return super().request_uri(identifier_str or "1970-01-01")
vlorentzUnsubmitted
Not Done
Inline Actions
last_update=last_update,
)
- def commit_page(self, page: List[Dict[str, Any]]):
+ def commit_page(self, page: List[Dict[str, Any]]) -> None:
"""Update the currently stored state using the latest listed page."""
vlorentz:
def commit_page(self, page: List[Dict[str, Any]]):
def is_within_bounds( """Update the currently stored state using the latest listed page"""
self, inner: int, lower: Optional[int] = None, upper: Optional[int] = None
) -> bool: last_repo = page[-1]
# values are expected to be datetimes last_repo_cdate = iso8601.parse_date(last_repo["created_on"])
if lower is None and upper is None:
ret = True if (
elif lower is None: self.state.last_repo_cdate is None
ret = inner <= upper # type: ignore or last_repo_cdate > self.state.last_repo_cdate
elif upper is None: ):
ret = inner >= lower self.state.last_repo_cdate = last_repo_cdate
else:
ret = lower <= inner <= upper def finalize(self):
return ret
scheduler_state = self.get_state_from_scheduler()
if self.state.last_repo_cdate is None:
return
# Update the lister state in the backend only if the last seen id of
# the current run is higher than that stored in the database.
if (
scheduler_state.last_repo_cdate is None
or self.state.last_repo_cdate > scheduler_state.last_repo_cdate
):
self.updated = True
About · Developer information · Legal