create-demo-comms.py

"""Demo communities and members creation.

Note: run ``pip install faker`` to use
"""
import uuid

from faker import Faker
from flask import current_app
from flask_security.utils import hash_password
from invenio_db import db
from invenio_pidstore import current_pidstore
from invenio_search import current_search
from werkzeug.local import LocalProxy

from invenio_communities.api import Community
from invenio_communities.indexer import CommunityIndexer
from invenio_communities.marshmallow import CommunitySchemaMetadataV1
from invenio_communities.members.models import CommunityMember, MembershipRequest
from invenio_communities.models import CommunityMetadata

_datastore = LocalProxy(lambda: current_app.extensions["security"].datastore)

FAKE = Faker()


def create_fake_community():
    """Create communities for demo purposes."""
    data = {
        "id": FAKE.slug(),
        "title": FAKE.sentence(),
        "description": FAKE.text(),
        "page": FAKE.text(),
        "type": FAKE.word(ext_word_list=["organization", "event", "topic", "project"]),
        "visibility": FAKE.word(ext_word_list=["public", "private", "hidden"]),
    }

    data = CommunitySchemaMetadataV1().load(data).data

    comm_uuid = uuid.uuid4()
    comid = current_pidstore.minters["comid"](comm_uuid, data)
    community = Community.create(data, id_=comm_uuid)
    db.session.commit()

    CommunityIndexer().index(community)

    return comid, community


def create_fake_member(community=None):
    """Create members for demo purposes."""
    community = community or CommunityMetadata.query.order_by(func.random()).first()

    data = {
        "email": FAKE.email(),
        "password": "123456",
        "active": True,
        "profile": {"full_name": FAKE.name(), "username": FAKE.slug()},
    }
    data["password"] = hash_password(data["password"])
    user = _datastore.create_user(**data)
    db.session.commit()

    comm_mem = CommunityMembers.create(
        comm_id=community.id,
        user_id=user.id,
        role=FAKE.word(ext_word_list=["A", "C", "M"]),
    )
    db.session.commit()

    return user, comm_mem


def create_demo_communities(n=10):
    for _ in range(n):
        comid, comm = create_fake_community()
        print(comid, comm)
    current_search.flush_and_refresh(index="communities")


def create_demo_members(n=10):
    for _ in range(n):
        user, comm_mem = create_fake_member()
        print(comid, comm)