How to use the clld.db.meta.DBSession function in clld

def test_compute_language_sources(data):
    from clld.db.util import compute_language_sources
    from clld.db.models.common import Source, Sentence, Language, SentenceReference
    from clld.db.meta import DBSession

    s = Sentence(id='sentenced', language=Language(id='newlang'))
    sr = SentenceReference(sentence=s, source=Source.first())
    DBSession.add(sr)
    DBSession.flush()
    compute_language_sources()

def test_source2source(env):
    from clld.web.adapters.cldf import source2source

    for source in DBSession.query(Source):
        res = source2source(env['request'], source)
        assert len(res) >= 1

def test_CsvMixin(db):
    l1 = Language(id='abc', name='Name', latitude=12.4, jsondata=dict(a=None))
    DBSession.add(l1)
    DBSession.flush()
    l1 = Language.csv_query(DBSession).first()
    cols = l1.csv_head()
    row = l1.to_csv()
    for k, v in zip(cols, row):
        if k == 'jsondata':
            assert 'a' in json.loads(v)
    l2 = Language.from_csv(row)
    assert pytest.approx(l1.latitude) == l2.latitude
    row[cols.index('latitude')] = '3,5'
    l2 = Language.from_csv(row)
    assert l2.latitude < l1.latitude

def testdb(mocker):
    subprocess.check_call(['dropdb', '-U', 'postgres', '--if-exists', TEST_DB])
    subprocess.check_call(['createdb', '-U', 'postgres', TEST_DB])
    engine = create_engine('postgresql://postgres@/' + TEST_DB)
    DBSession.configure(bind=engine)
    Base.metadata.create_all(engine)

    mocker.patch('glottolog3.__main__.db_url', mocker.Mock(return_value=engine.url))
    mocker.patch('glottolog3.__main__.with_session')

    yield DBSession
    DBSession.close_all()
    DBSession.remove()
    engine.dispose()
    subprocess.check_call(['dropdb', '-U', 'postgres', '--if-exists', TEST_DB])

kw.update(
        publisher=entry.publisher_and_address[0],
        address=entry.publisher_and_address[1],
        year_int=entry.year_int,
        pages_int=entry.pages_int,
        med_index=-entry.weight[0],
        med_pages=entry.weight[1],
        med_type=entry.med_type.id,
        id=entry.fields['glottolog_ref_id'],
        fts=fts.tsvector('\n'.join(v for k, v in entry.fields.items() if k != 'abstract')),
        name='{} {}'.format(entry.fields.get('author', 'na'), entry.fields.get('year', 'nd')),
        description=entry.fields.get('title') or entry.fields.get('booktitle'),
        bibtex_type=btype)
    ref = models.Ref(**kw)
    DBSession.add(ref)
    DBSession.flush()

    reflangs, trigger = [], None
    no_ca = [{'degruyter'}, {'benjamins'}]
    provs = set()
    for key in entry.fields['srctrickle'].split(','):
        key = key.strip()
        if key:
            reflangs.extend(lgsources.get(key, []))
            prov, key = key.split('#', 1)
            provs.add(prov)
            DBSession.add(models.Refprovider(
                provider_pk=data['Provider'][prov].pk,
                ref_pk=ref.pk,
                id='{0}:{1}'.format(prov, key)))

    if not reflangs:

def query(cls):
        return DBSession.query(Parameter)

parser.add_argument(
        "config_uri", action=ExistingConfig, help="ini file providing app config")
    parser.add_argument("--glottolog-dburi", default=None)
    parser.add_argument("--module", default=None)
    parser.add_argument(
        "--sqlite", nargs=1, action=SqliteDb, help="sqlite db file")
    for args, _kw in arg_specs:
        parser.add_argument(*args, **_kw)
    args = parser.parse_args(args=kw.pop('args', None))
    engine = getattr(args, 'engine', kw.get('engine', None))
    args.env = bootstrap(args.config_uri) if kw.get('bootstrap', False) else {}
    module = setup_session(args.config_uri, engine=engine)

    # make sure we create URLs in the correct domain
    if args.env:
        dataset = DBSession.query(common.Dataset).first()
        if dataset:
            args.env['request'].environ['HTTP_HOST'] = dataset.domain

    if module == 'tests':
        module = 'clld'
    args.module = __import__(args.module or module)
    args.log = logging.getLogger(args.module.__name__)
    if engine:
        args.log.info('using bind %s' % engine)
    args.data_file = partial(data_file, args.module)
    args.module_dir = Path(args.module.__file__).parent
    args.migrations_dir = Path(args.module.__file__).parent.joinpath('..', 'migrations')
    return args

common.LanguageIdentifier,
            common.Identifier, and_(
                common.LanguageIdentifier.identifier_pk == common.Identifier.pk,
                common.Identifier.type != 'name')
        )).filter(common.Language.active == true()).order_by(common.Language.id)

        def resources():
            for (id, name, lat, lon), rows in groupby(q, itemgetter(0, 1, 2, 3)):
                identifiers = [
                    {'type': r.itype, 'identifier': r.iname.lower()
                     if r.itype.startswith('WALS') else r.iname}
                    for r in rows if r.iname is not None]
                yield {'id': id, 'name': name, 'latitude': lat, 'longitude': lon,
                       'identifiers': identifiers}
    elif rsc == 'parameter':
        q = DBSession.query(
            common.Parameter.id,
            common.Parameter.name
        ).order_by(common.Parameter.pk)

        def resources():
            for id, name in q:
                yield {'id': id, 'name': name}
    else:
        return HTTPNotFound()

    return {
        'properties': {
            'dataset': req.dataset.id,
            'uri_template': get_url_template(req, rsc, relative=False)},
        'resources': list(resources())}

def get_source(key):
    return DBSession.query(Ref).join(Refprovider).filter(Refprovider.id == key).one()