# Licensed to the Apache Software Foundation (ASF) under one
# or more contributor license agreements. See the NOTICE file
# distributed with this work for additional information
# regarding copyright ownership. The ASF licenses this file
# to you under the Apache License, Version 2.0 (the
# "License"); you may not use this file except in compliance
# with the License. You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing,
# software distributed under the License is distributed on an
# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
# KIND, either express or implied. See the License for the
# specific language governing permissions and limitations
# under the License.
from io import BytesIO
import datetime
import string
import logging
try:
from typing import Union
except ImportError: # py2 doesn't have typing yet
pass
import colander as col
import bson
import requests
import formencode
import six
from urllib.parse import urlparse
from allura.lib.helpers import slugify
from allura.model import Neighborhood
from ming.base import Object
from ming.odm import ThreadLocalODMSession
from tg import tmpl_context as c, app_globals as g
from allura import model as M
from allura.lib import helpers as h
from allura.lib.plugin import ProjectRegistrationProvider
log = logging.getLogger(__name__)
"""
Validators and helpers for adding projects via the API or project-import.py script
"""
class TroveCategory():
def __init__(self, root_type=''):
self.root_type = root_type
def deserialize(self, node, cstruct):
if cstruct is col.null:
return col.null
cat = M.TroveCategory.query.get(fullpath=cstruct)
if not cat:
cat = M.TroveCategory.query.get(fullname=cstruct)
if not cat:
raise col.Invalid(node,
'"%s" is not a valid trove category.' % cstruct)
if not cat.fullpath.startswith(self.root_type):
raise col.Invalid(node,
'"%s" is not a valid "%s" trove category.' %
(cstruct, self.root_type))
return cat
class User():
def deserialize(self, node, cstruct):
if cstruct is col.null:
return col.null
user = M.User.by_username(cstruct)
if not user:
raise col.Invalid(node,
'Invalid username "%s".' % cstruct)
return user
class ProjectShortnameType():
def __init__(self, nbhd, update):
self.nbhd = nbhd
self.update = update
def deserialize(self, node, cstruct):
if cstruct is col.null:
return col.null
return ProjectRegistrationProvider.get().shortname_validator.to_python(cstruct,
check_allowed=not self.update,
neighborhood=self.nbhd)
class Award():
def __init__(self, nbhd):
self.nbhd = nbhd
def deserialize(self, node, cstruct):
if cstruct is col.null:
return col.null
award = M.Award.query.find(dict(short=cstruct,
created_by_neighborhood_id=self.nbhd._id)).first()
if not award:
# try to look up the award by _id
award = M.Award.query.find(dict(_id=bson.ObjectId(cstruct),
created_by_neighborhood_id=self.nbhd._id)).first()
if not award:
raise col.Invalid(node,
'Invalid award "%s".' % cstruct)
return award
class TroveTopics(col.SequenceSchema):
trove_topics = col.SchemaNode(TroveCategory("Topic"))
class TroveLicenses(col.SequenceSchema):
trove_license = col.SchemaNode(TroveCategory("License"))
class TroveDatabases(col.SequenceSchema):
trove_databases = col.SchemaNode(TroveCategory("Database Environment"))
class TroveStatuses(col.SequenceSchema):
trove_statuses = col.SchemaNode(TroveCategory("Development Status"))
class TroveAudiences(col.SequenceSchema):
trove_audience = col.SchemaNode(TroveCategory("Intended Audience"))
class TroveOSes(col.SequenceSchema):
trove_oses = col.SchemaNode(TroveCategory("Operating System"))
class TroveLanguages(col.SequenceSchema):
trove_languages = col.SchemaNode(TroveCategory("Programming Language"))
class TroveTranslations(col.SequenceSchema):
trove_translations = col.SchemaNode(TroveCategory("Translations"))
class TroveUIs(col.SequenceSchema):
trove_uis = col.SchemaNode(TroveCategory("User Interface"))
class Labels(col.SequenceSchema):
label = col.SchemaNode(col.Str())
class NewProjectSchema(col.MappingSchema):
def schema_type(self, **kw):
return col.Mapping(unknown='raise')
name = col.SchemaNode(col.Str())
summary = col.SchemaNode(col.Str(), missing='')
description = col.SchemaNode(col.Str(), missing='')
admin = col.SchemaNode(User())
private = col.SchemaNode(col.Bool(), missing=False)
labels = Labels(missing=[])
external_homepage = col.SchemaNode(col.Str(), missing='')
video_url = col.SchemaNode(col.Str(), missing='')
trove_root_databases = TroveDatabases(missing=None)
trove_developmentstatuses = TroveStatuses(validator=col.Length(max=6), missing=None)
trove_audiences = TroveAudiences(validator=col.Length(max=6), missing=None)
trove_licenses = TroveLicenses(validator=col.Length(max=6), missing=None)
trove_oses = TroveOSes(missing=None)
trove_languages = TroveLanguages(validator=col.Length(max=6), missing=None)
trove_topics = TroveTopics(validator=col.Length(max=3), missing=None)
trove_natlanguages = TroveTranslations(missing=None)
trove_environments = TroveUIs(missing=None)
tool_data = col.SchemaNode(col.Mapping(unknown='preserve'), missing={})
icon_url = col.SchemaNode(col.Str(), missing=None, validator=col.url)
# more fields are added dynamically to the schema in make_newproject_schema()
def trove_ids(orig, new_):
if new_ is None:
return orig
return list({t._id for t in list(new_)})
def make_newproject_schema(nbhd, update=False):
# type: (Neighborhood, bool) -> NewProjectSchema
projectSchema = NewProjectSchema(unknown='raise')
# dynamically add to the schema fields that depend on `nbhd`
projectSchema.add(col.SchemaNode(col.Sequence(),
col.SchemaNode(Award(nbhd)),
name='awards', missing=[]))
projectSchema.add(col.SchemaNode(ProjectShortnameType(nbhd, update),
name='shortname', missing=None))
return projectSchema
def deserialize_project(datum, projectSchema, nbhd):
# type: (dict, NewProjectSchema, Neighborhood) -> object
p = projectSchema.deserialize(datum)
p = Object(p) # convert from dict to something with attr-access
# generate a shortname, and try to make it unique
if not p.shortname:
max_shortname_len = 15 # maybe more depending on NeighborhoodProjectShortNameValidator impl, but this is safe
shortname = orig_shortname = make_shortname(p.name, max_shortname_len)
for i in range(1, 10):
try:
ProjectRegistrationProvider.get().shortname_validator.to_python(shortname, neighborhood=nbhd)
except formencode.api.Invalid:
if len(orig_shortname) == max_shortname_len - 1:
shortname = orig_shortname + str(i)
else:
shortname = orig_shortname[:max_shortname_len - 1] + str(i)
else:
# we're good!
break
p.shortname = shortname
return p
def make_shortname(name, max_len):
# lowercase, drop periods and underscores
shortname = slugify(name)[1].replace('_', '-')
# must start with a letter
if not shortname[0].isalpha():
shortname = 'a-' + shortname
# truncate length, avoid trailing dash
shortname = shortname[:max_len].rstrip('-')
# too short
if len(shortname) < 3:
shortname += '-z'
return shortname
def create_project_with_attrs(p, nbhd, update=False, ensure_tools=False):
# type: (object, M.Neighborhood, bool, bool) -> Union[M.Project|bool]
M.session.artifact_orm_session._get().skip_mod_date = True
shortname = p.shortname
project = M.Project.query.get(shortname=shortname,
neighborhood_id=nbhd._id)
project_template = nbhd.get_project_template()
if project and not update:
log.warning('Skipping existing project "%s"' % (shortname))
return False
if not project:
creating = True
project = nbhd.register_project(shortname,
p.admin,
project_name=p.name,
private_project=p.private,
omit_event=True, # because we'll fire it later after setting other fields
)
else:
creating = False
log.info('Updating project "%s".' % (shortname))
project.notifications_disabled = True
if ensure_tools and 'tools' in project_template:
for i, tool in enumerate(project_template['tools'].keys()):
tool_config = project_template['tools'][tool]
if project.app_instance(tool_config['mount_point']):
continue
tool_options = tool_config.get('options', {})
for k, v in list(tool_options.items()):
if isinstance(v, str):
tool_options[k] = string.Template(v).safe_substitute(
project.root_project.__dict__.get('root_project', {}))
project.install_app(tool,
mount_label=tool_config['label'],
mount_point=tool_config['mount_point'],
**tool_options)
project.summary = p.summary
project.short_description = p.description
project.external_homepage = p.external_homepage
project.video_url = p.video_url
project.last_updated = datetime.datetime.utcnow()
# These properties may have been populated by nbhd template defaults in
# register_project(). Overwrite if we have data, otherwise keep defaults.
project.labels = p.labels or project.labels
project.trove_root_database = trove_ids(project.trove_root_database, p.trove_root_databases)
project.trove_developmentstatus = trove_ids(project.trove_developmentstatus, p.trove_developmentstatuses)
project.trove_audience = trove_ids(project.trove_audience, p.trove_audiences)
project.trove_license = trove_ids(project.trove_license, p.trove_licenses)
project.trove_os = trove_ids(project.trove_os, p.trove_oses)
project.trove_language = trove_ids(project.trove_language, p.trove_languages)
project.trove_topic = trove_ids(project.trove_topic, p.trove_topics)
project.trove_natlanguage = trove_ids(project.trove_natlanguage, p.trove_natlanguages)
project.trove_environment = trove_ids(project.trove_environment, p.trove_environments)
project.tool_data.update(p.tool_data)
for a in p.awards:
M.AwardGrant(app_config_id=bson.ObjectId(),
award_id=a._id,
granted_to_project_id=project._id,
granted_by_neighborhood_id=nbhd._id)
if p.icon_url:
req = requests.get(p.icon_url, timeout=30)
req.raise_for_status()
project.save_icon(urlparse(p.icon_url).path,
BytesIO(req.content),
content_type=req.headers.get('Content-Type'))
elif getattr(p, 'icon', None):
with open(p.icon, 'rb') as icon_file:
project.save_icon(p.icon, icon_file)
project.notifications_disabled = False
with h.push_config(c, project=project, user=p.admin):
ThreadLocalODMSession.flush_all()
if creating:
g.post_event('project_created')
else:
g.post_event('project_updated')
return project