Differential D7879 Diff 28433 swh/lister/maven/lister.py

Changeset View

Standalone View

View Options

swh/lister/maven/lister.py

# See the AUTHORS file at the top-level directory of this distribution

# License: GNU General Public License version 3, or any later version

# See top-level LICENSE file for more information

from dataclasses import asdict, dataclass

from datetime import datetime, timezone

import logging

import re

from typing import Any, Dict, Iterator, Optional

from urllib.parse import urljoin

import requests

from tenacity.before_sleep import before_sleep_log

import xmltodict

from swh.core.github.utils import GitHubSession

from swh.lister.utils import throttling_retry

from swh.scheduler.interface import SchedulerInterface

from swh.scheduler.model import ListedOrigin

from .. import USER_AGENT

from ..pattern import CredentialsType, Lister

logger = logging.getLogger(__name__)

Show All 22 Lines

class MavenLister(Lister[MavenListerState, RepoPage]):

Maven Central provides artifacts for Java builds.

It includes POM files and source archives, which we download to get

the source code of artifacts and links to their scm repository.

This lister yields origins of types: git/svn/hg or whatever the Artifacts

use as repository type, plus maven types for the maven loader (tgz, jar)."""

LISTER_NAME = "maven"

vlorentzUnsubmitted

Not Done

could you move it to the toplevel? it doesn't need to be an attribute

vlorentz: could you move it to the toplevel? it doesn't need to be an attribute

ardumontAuthorUnsubmitted

Done

sure

ardumont: sure

SUPPORTED_SCM_TYPES = ("git", "svn", "hg", "cvs", "bzr")

def __init__(

self,

scheduler: SchedulerInterface,

url: str,

index_url: str = None,

instance: Optional[str] = None,

credentials: CredentialsType = None,

Show All 29 Lines

self.session.headers.update(

{

"Accept": "application/json",

"User-Agent": USER_AGENT,

}

)

self.jar_origins: Dict[str, ListedOrigin] = {}

self.github_session = GitHubSession(

credentials=self.credentials, user_agent=USER_AGENT

)

def state_from_dict(self, d: Dict[str, Any]) -> MavenListerState:

return MavenListerState(**d)

def state_to_dict(self, state: MavenListerState) -> Dict[str, Any]:

return asdict(state)

@throttling_retry(before_sleep=before_sleep_log(logger, logging.WARNING))

▲ Show 20 Lines • Show All 157 Lines • ▼ Show 20 Lines

def get_pages(self) -> Iterator[RepoPage]:

except requests.HTTPError:

logger.warning(

"POM info page could not be fetched, skipping project '%s'",

pom,

)

except xmltodict.expat.ExpatError as error:

logger.info("Could not parse POM %s XML: %s. Next.", pom, error)

def get_origins_from_page(self, page: RepoPage) -> Iterator[ListedOrigin]:

def get_scm(self, page: RepoPage) -> Optional[ListedOrigin]:

"""Convert a page of Maven repositories into a list of ListedOrigins."""

"""Retrieve scm origin out of the page information. Only called when type of the

assert self.lister_obj.id is not None

page is scm.

scm_types_ok = ("git", "svn", "hg", "cvs", "bzr")

if page["type"] == "scm":

Try and detect an scm. Note that official format is of the form:

# If origin is a scm url: detect scm type and yield.

scm:{type}:git://example.org/{user}/{repo}.git but some projects directly put

# Note that the official format is:

the repo url (without the "scm:type"), so we have to check against the content

# scm:git:git://github.com/openengsb/openengsb-framework.git

to extract the type, url properly.

# but many, many projects directly put the repo url, so we have to

# detect the content to match it properly.

Raises

AssertionError when page['type'] != 'scm'

Returns

ListedOrigin with proper canonical scm url (for github) if any is found,

None otherwise.

"""

assert page["type"] == "scm"

visit_type: Optional[str] = None

url: Optional[str] = None

origin: Optional[ListedOrigin] = None

m_scm = re.match(r"^scm:(?P<type>[^:]+):(?P<url>.*)$", page["url"])

if m_scm is not None:

scm_type = m_scm.group("type")

scm_type = m_scm.group("type") if m_scm is not None else None

if scm_type in scm_types_ok:

if m_scm and scm_type and scm_type in self.SUPPORTED_SCM_TYPES:

scm_url = m_scm.group("url")

url = m_scm.group("url")

origin = ListedOrigin(

visit_type = scm_type

lister_id=self.lister_obj.id,

elif page["url"].endswith(".git"):

url=scm_url,

url = page["url"]

visit_type=scm_type,

visit_type = "git"

)

yield origin

if url and visit_type and visit_type == "git":

else:

url = self.github_session.get_canonical_url(url)

if page["url"].endswith(".git"):

if url and visit_type:

assert self.lister_obj.id is not None

origin = ListedOrigin(

lister_id=self.lister_obj.id,

url=page["url"],

url=url,

visit_type="git",

visit_type=visit_type,

vlorentzUnsubmitted

Not Done

return None

- if url and visit_type == "git":

+ if not url:

+ return None

+ if visit_type == "git":

# Non-github urls will be returned as is, github ones will be canonical ones

url = self.github_session.get_canonical_url(url)

- if not url:

- return None

assert visit_type is not None

another one

vlorentz: another one

ardumontAuthorUnsubmitted

Done

No, here it must be after the get_canonical call because that could return None.
Hence why i put it there

ardumont: No, here it must be after the get_canonical call because that could return None. Hence why i…

vlorentzUnsubmitted

Not Done

ack

vlorentz: ack

)

yield origin

return origin

vlorentzUnsubmitted

Not Done

m_scm = re.match(r"^scm:(?P<type>[^:]+):(?P<url>.*)$", page["url"])

- scm_type = m_scm.group("type") if m_scm is not None else None

- if m_scm and scm_type and scm_type in self.SUPPORTED_SCM_TYPES:

+ if m_scm is None:

+ return None

+ scm_type = m_scm.group("type")

+ if scm_type and scm_type in self.SUPPORTED_SCM_TYPES:

url = m_scm.group("url")

visit_type = scm_type

elif page["url"].endswith(".git"):

url = page["url"].lstrip("scm:")

visit_type = "git"

+ else:

+ return None

- if url and visit_type and visit_type == "git":

+ if visit_type == "git":

# Non-github urls will be returned as is, github ones will be canonical ones

url = self.github_session.get_canonical_url(url)

- if url and visit_type:

- assert self.lister_obj.id is not None

- origin = ListedOrigin(

- lister_id=self.lister_obj.id,

- url=url,

- visit_type=visit_type,

- )

- return origin

+ assert self.lister_obj.id is not None

+ return ListedOrigin(

+ lister_id=self.lister_obj.id,

+ url=url,

+ visit_type=visit_type,

+ )

def get_origins_from_page(self, page: RepoPage) -> Iterator[ListedOrigin]:

It is simpler to return early than to re-check the same conditions every time, IMO

vlorentz: It is simpler to return early than to re-check the same conditions every time, IMO

ardumontAuthorUnsubmitted

Done

yeah, i was missing the else return None you put early on, thanks.

ardumont: yeah, i was missing the else return None you put early on, thanks.

def get_origins_from_page(self, page: RepoPage) -> Iterator[ListedOrigin]:

"""Convert a page of Maven repositories into a list of ListedOrigins."""

if page["type"] == "scm":

listed_origin = self.get_scm(page)

if listed_origin:

yield listed_origin

else:

# Origin is gathering source archives:

last_update_dt = None

last_update_iso = ""

try:

last_update_seconds = str(page["time"])[:-3]

last_update_dt = datetime.fromtimestamp(int(last_update_seconds))

last_update_dt = last_update_dt.astimezone(timezone.utc)

Show All 10 Lines

def get_origins_from_page(self, page: RepoPage) -> Iterator[ListedOrigin]:

artifact = {

**{k: v for k, v in page.items() if k != "doc"},

"time": last_update_iso,

"base_url": self.BASE_URL,

}

if origin_url not in self.jar_origins:

# Create ListedOrigin instance if we did not see that origin yet

assert self.lister_obj.id is not None

jar_origin = ListedOrigin(

lister_id=self.lister_obj.id,

url=origin_url,

visit_type=page["type"],

last_update=last_update_dt,

extra_loader_arguments={"artifacts": [artifact]},

)

self.jar_origins[origin_url] = jar_origin

▲ Show 20 Lines • Show All 54 Lines • Show Last 20 Lines