feat: add seeding data endpoint

alembic.ini

@@ -60,7 +60,7 @@ version_path_separator = os  # Use os.pathsep. Default configuration used for ne
 # are written from script.py.mako
 # output_encoding = utf-8
 
-sqlalchemy.url = 
+sqlalchemy.url = postgresql://username:password@localhost:5432/test
 
 
 [post_write_hooks]

api/utils/db_utils.py

@@ -0,0 +1,58 @@
+# api/utils/db_utils.py
+from sqlalchemy.orm import Session
+from api.v1.models.user import User
+from api.v1.models.organisation import Organisation
+from api.v1.models.profile import Profile
+from api.v1.models.permissions.user_org_role import user_organisation_roles
+from passlib.context import CryptContext
+
+pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
+
+def hash_password(password: str):
+    return pwd_context.hash(password)
+
+def seed_user_and_org(db: Session, user_data: dict):
+    # Seed user
+    hashed_password = hash_password(user_data['password'])
+    user = User(
+        email=user_data['email'],
+        password=hashed_password,
+        first_name=user_data.get('first_name'),
+        last_name=user_data.get('last_name'),
+        avatar_url=user_data.get('avatar_url'),
+        is_superadmin=user_data.get('is_superadmin', False),
+        username=user_data.get('username') #add username to user table
+    )
+    db.add(user)
+    db.commit()
+
+    # Seed organization
+    org = Organisation(name=user_data['org_name'], email=user_data['org_email'])
+    db.add(org)
+    db.commit()
+
+    # Seed profile
+    profile = Profile(
+        user_id=user.id,
+        username=user_data.get('username'),
+        pronouns=user_data.get('pronouns'),
+        job_title=user_data.get('job_title'),
+        department=user_data.get('department'),
+        social=user_data.get('social'),
+        bio=user_data.get('bio'),
+        phone_number=user_data.get('phone_number'),
+        avatar_url=user_data.get('profile_avatar_url'),
+        recovery_email=user_data.get('recovery_email'),
+        facebook_link=user_data.get('facebook_link'),
+        instagram_link=user_data.get('instagram_link'),
+        twitter_link=user_data.get('twitter_link'),
+        linkedin_link=user_data.get('linkedin_link'),
+    )
+    db.add(profile)
+    db.commit()
+
+    # Seed association
+    db.execute(user_organisation_roles.insert().values(user_id=user.id, organisation_id=org.id, is_owner=True, status='active'))
+    db.commit()
+
+    return user, org, profile

api/v1/models/seed.py

@@ -0,0 +1,29 @@
+from api.v1.models.user import User
+from api.v1.models.organisation import Organisation
+from api.v1.models.permissions.user_org_role import user_organisation_roles
+from sqlalchemy.orm import Session
+from faker import Faker
+
+fake = Faker()
+
+def seed_user_and_org(db: Session):
+    # Create a user
+    user = User(email=fake.email(), password=fake.password(), first_name=fake.first_name(), last_name=fake.last_name())
+    db.add(user)
+    db.commit()
+
+    # Create an organization
+    org = Organisation(name=fake.company(), email=fake.company_email())
+    db.add(org)
+    db.commit()
+
+    # Link the user and organization in the association table
+    db.execute(user_organisation_roles.insert().values(
+        user_id=user.id,
+        organisation_id=org.id,
+        is_owner=True, #make the user the owner.
+        status='active'
+    ))
+    db.commit()
+
+    return user, org

api/v1/models/user.py

@@ -20,6 +20,7 @@ class User(BaseTableModel):
     is_superadmin = Column(Boolean, server_default=text("false"))
     is_deleted = Column(Boolean, server_default=text("false"))
     is_verified = Column(Boolean, server_default=text("false"))
+    username = Column(String, nullable=True)
 
     profile = relationship(
         "Profile", uselist=False, back_populates="user", cascade="all, delete-orphan"

api/v1/routes/__init__.py

@@ -46,6 +46,7 @@
 from api.v1.routes.settings import settings
 from api.v1.routes.terms_and_conditions import terms_and_conditions
 from api.v1.routes.stripe import subscription_
+from api.v1.routes.seed import seed
 
 api_version_one = APIRouter(prefix="/api/v1")
 
@@ -96,3 +97,4 @@
 api_version_one.include_router(terms_and_conditions)
 api_version_one.include_router(product_comment)
 api_version_one.include_router(subscription_)
+api_version_one.include_router(seed)

api/v1/routes/seed.py

@@ -0,0 +1,61 @@
+# api/v1/routes/seed.py
+from fastapi import APIRouter, Depends, Query, status, HTTPException
+from sqlalchemy.orm import Session
+from api.db.database import get_db
+from api.utils.db_utils import seed_user_and_org
+from faker import Faker
+
+fake = Faker()
+
+seed = APIRouter(tags=['Seed'])
+
+@seed.post("/seed")
+def seed_users(db: Session = Depends(get_db), num_users: int = Query(1, description="Number of fake users to create")):
+    """
+    Seeds the database with fake users and organizations.
+    """
+    try:
+        users_created = []
+        for _ in range(num_users):
+            user_data = {
+                'email': fake.email(),
+                'password': 'testpassword',
+                'first_name': fake.first_name(),
+                'last_name': fake.last_name(),
+                'avatar_url': fake.image_url(),
+                'is_superadmin': fake.boolean(),
+                'org_name': fake.company(),
+                'org_email': fake.company_email(),
+                'username': fake.user_name(),
+                'pronouns': fake.random_element(elements=('they/them', 'he/him', 'she/her')),
+                'job_title': fake.job(),
+                'department': fake.random_element(elements=('Engineering', 'Sales', 'Marketing')),
+                'social': '{"github": "' + fake.user_name() + '"}',
+                'bio': fake.text(),
+                'phone_number': fake.phone_number(),
+                'profile_avatar_url': fake.image_url(),
+                'recovery_email': fake.email(),
+                'facebook_link': fake.url(),
+                'instagram_link': fake.url(),
+                'twitter_link': fake.url(),
+                'linkedin_link': fake.url(),
+            }
+            user, org, profile = seed_user_and_org(db, user_data)
+            users_created.append(user.email)
+
+        return {
+            "status": "success",
+            "status_code": status.HTTP_200_OK,
+            "message": f"{num_users} users and organizations seeded successfully.",
+            "data": {"users": users_created}
+        }
+    except Exception as e:
+        raise HTTPException(
+            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
+            detail={
+                "status": "error",
+                "status_code": status.HTTP_500_INTERNAL_SERVER_ERROR,
+                "message": f"An error occurred: {str(e)}",
+                "data": None
+            }
+        )

api/v1/schemas/seed.py

@@ -0,0 +1,14 @@
+# api/v1/schemas/seed.py
+from pydantic import BaseModel
+from typing import Optional
+
+class SeedUserRequest(BaseModel):
+    email: str
+    password: str
+    first_name: Optional[str] = None
+    last_name: Optional[str] = None
+    is_active: Optional[bool] = True
+    is_superadmin: Optional[bool] = False
+    is_deleted: Optional[bool] = False
+    is_verified: Optional[bool] = False
+    organization_name: Optional[str] = None #Add organization name

tests/test_seed.py

@@ -0,0 +1,68 @@
+import pytest
+from fastapi.testclient import TestClient
+from sqlalchemy import create_engine, text
+from sqlalchemy.orm import sessionmaker
+
+from main import app
+from api.db.database import get_db, Base
+
+# PostgreSQL connection string for your test database.
+SQLALCHEMY_DATABASE_URL = "postgresql://username:password@localhost:5432/test"
+
+engine = create_engine(SQLALCHEMY_DATABASE_URL)
+TestingSessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
+
+def reset_test_schema(engine, schema_name="test_schema"):
+    with engine.connect() as connection:
+        connection.execute(text(f"DROP SCHEMA IF EXISTS {schema_name} CASCADE"))
+        connection.execute(text(f"CREATE SCHEMA {schema_name}"))
+        connection.execute(text(f"SET search_path TO {schema_name}"))
+        connection.commit()
+
+reset_test_schema(engine, schema_name="test_schema")
+
+# Ensure that the Base metadata uses the test schema.
+Base.metadata.schema = "test_schema"
+Base.metadata.create_all(bind=engine)
+
+def override_get_db():
+    db = TestingSessionLocal()
+    try:
+        yield db
+    finally:
+        db.close()
+
+app.dependency_overrides[get_db] = override_get_db
+
+client = TestClient(app)
+
+def test_seed_endpoint():
+    response = client.post("/api/v1/seed", params={"num_users": 2})
+    assert response.status_code == 200, response.text
+    data = response.json()
+    assert data["status"] == "success"
+    assert data["status_code"] == 200
+    assert data["message"].startswith("2 users and organizations seeded successfully.")
+    assert len(data["data"]["users"]) == 2
+
+def test_seed_endpoint_non_integer_num_users():
+    response = client.post("/api/v1/seed", params={"num_users": "abc"})
+    assert response.status_code == 422, response.text
+
+def test_seed_endpoint_no_num_users():
+    response = client.post("/api/v1/seed")
+    assert response.status_code == 200, response.text
+    data = response.json()
+    assert data["status"] == "success"
+    assert data["status_code"] == 200
+    assert data["message"].startswith("1 users and organizations seeded successfully.")
+    assert len(data["data"]["users"]) == 1
+
+def test_seed_endpoint_large_num_users():
+    response = client.post("/api/v1/seed", params={"num_users": 100})
+    assert response.status_code == 200, response.text
+    data = response.json()
+    assert data["status"] == "success"
+    assert data["status_code"] == 200
+    assert data["message"].startswith("100 users and organizations seeded successfully.")
+    assert len(data["data"]["users"]) == 100