Skip to content

Instantly share code, notes, and snippets.

@vsajip
Last active March 12, 2017 23:06
Show Gist options
  • Star 0 You must be signed in to star a gist
  • Fork 1 You must be signed in to fork a gist
  • Save vsajip/5929707 to your computer and use it in GitHub Desktop.
Save vsajip/5929707 to your computer and use it in GitHub Desktop.
Simple RDBMS modelling of dependencies based on PEP 426. Needs SQLAlchemy 0.8. Now updated to reflect the latest schema from PEP 426.
import codecs
import itertools
import json
import optparse # support 2.6
import os
import sys
from sqlalchemy import create_engine
from sqlalchemy import Column, Date, Integer, String, ForeignKey
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, scoped_session, relationship, backref
from sqlalchemy.sql.expression import ClauseElement
DBNAME = 'pypi.db'
ECHO = False
engine = create_engine('sqlite:///' + DBNAME, echo=ECHO)
Session = scoped_session(sessionmaker(bind=engine))
session = Session()
Base = declarative_base()
class Entity(Base):
__abstract__ = True
id = Column(Integer, primary_key=True)
class Project(Entity):
__tablename__ = 'projects'
name = Column(String)
releases = relationship('Release', backref='project',
cascade='all,delete,delete-orphan')
class Release(Entity):
__tablename__ = 'releases'
version = Column(String)
project_id = Column(Integer, ForeignKey('projects.id'))
#project = relationship(Project, backref=backref('releases'),
# primaryjoin=project_id == Project.id)
dependencies = relationship('Dependency', backref='release',
cascade='all,delete,delete-orphan')
class Dependency(Entity):
__tablename__ = 'dependencies'
release_id = Column(Integer, ForeignKey('releases.id'))
#release = relationship(Release, backref=backref("dependencies"),
# primaryjoin=release_id == Release.id)
install = Column(String, nullable=False)
extra = Column(String)
environment = Column(String)
kind = Column(String)
def get_or_create(session, model, **kwargs):
instance = session.query(model).filter_by(**kwargs).first()
if instance:
return instance, False
else:
instance = model(**kwargs)
session.add(instance)
return instance, True
def do_import(filename):
with codecs.open(filename, 'r', encoding='utf-8') as f:
data = json.load(f)
project, _ = get_or_create(session, Project,
name=data['name'])
release, _ = get_or_create(session, Release, project=project,
version=data['version'])
# Wipe existing dependencies
release.dependencies[:] = []
for key in ('build', 'dev', 'meta', 'run', 'test'):
kind = ':%s:' % key
k = '%s_requires' % key
if k in data:
for info in data[k]:
extra = info.get('extra')
environment = info.get('environment')
for r in info['requires']:
dep = Dependency(release=release,
install=r,
extra=extra,
environment=environment,
kind=kind)
release.dependencies.append(dep)
session.commit()
def do_export(proj_ver):
name, version = proj_ver.split('/', 1)
project = session.query(Project).filter_by(name=name).first()
if project is None:
print('No such project: %r' % name)
return
release = session.query(Release).filter_by(project=project,
version=version).first()
if release is None:
print('Project %r has no version %r' % (name, version))
return
result = {
'name': name,
'version': version,
}
def criteria(dep):
return dep.kind, dep.extra, dep.environment
deps = sorted(release.dependencies, key=criteria)
extras = set()
for k, g in itertools.groupby(deps, criteria):
kind, extra, environment = k
k = kind[1:-1]
g = list(g)
rlist = [e.install for e in g]
key = '%s_requires' % k
info = {
'requires': rlist
}
if extra:
info['extra'] = extra
extras.add(extra)
if environment:
info['environment'] = environment
result.setdefault(key, []).append(info)
if extras:
result['extras'] = sorted(extras)
json.dump(result, sys.stdout, ensure_ascii=False, indent=2)
def main():
parser = optparse.OptionParser()
parser.add_option('-r', dest='reset', default=False, action='store_true',
help='Clear and reinitialise database')
parser.add_option('-e', dest='exp', metavar='EXPORT',
help='Export named project/version as printed JSON')
parser.add_option('-i', dest='imp', metavar='IMPORT',
help='Import from JSON file')
options, args = parser.parse_args()
if options.reset:
os.remove(DBNAME)
Base.metadata.create_all(engine)
if options.imp:
do_import(options.imp)
elif options.exp:
do_export(options.exp)
else:
parser.print_help()
if __name__ == '__main__':
try:
rc = main()
except Exception as e:
print(e)
rc = 1
sys.exit(rc)
{
"name": "pyramid",
"run_requires": [
],
"test_requires": [
{
"requires": [
"WebTest (>= 1.3.1)",
"zope.component (>= 3.11.0)"
]
}
],
"version": "1.4.2",
"run_requires": [
{
"requires": [
"setuptools",
"Chameleon (>= 1.2.3)",
"Mako (>= 0.3.6)",
"WebOb (>= 1.2b3)",
"repoze.lru (>= 0.4)",
"zope.interface (>= 3.8.0)",
"zope.deprecation (>= 3.5.0)",
"venusian (>= 1.0a3)",
"translationstring (>= 0.4)",
"PasteDeploy (>= 1.5.0)"
]
},
{
"requires": [
"Sphinx",
"docutils",
"repoze.sphinx.autointerface"
],
"extra": "docs"
},
{
"requires": [
"WebTest (>= 1.3.1)",
"zope.component (>= 3.11.0)",
"nose",
"coverage",
"virtualenv"
],
"extra": "testing"
}
],
"extras": [
"docs",
"testing"
]
}
{
"version": "0.7.7",
"name": "setuptools",
"run_requires": [
{
"environment": "sys_platform=='win32'",
"requires": [
"wincertstore (== 0.1)"
],
"extra": "ssl"
},
{
"environment": "sys_platform=='win32' and python_version=='2.4'",
"requires": [
"ctypes (== 1.0.2)"
],
"extra": "ssl"
},
{
"requires": [
"certifi (== 0.0.8)"
],
"extra": "certs"
},
{
"environment": "python_version in '2.4, 2.5'",
"requires": [
"ssl (== 1.16)"
],
"extra": "ssl"
}
],
"extras": [
"certs",
"ssl"
]
}
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment