users-core

Minimal user identity management package following Hexagonal Architecture and Domain-Driven Design principles.

Philosophy

users-core manages the core identity of users in your system - nothing more, nothing less:

• UUID - Unique identifier
• Email - Unique, case-insensitive contact
• Username - Unique, case-insensitive, URL-safe identifier
• Status - Lifecycle state (active, suspended, archived, deleted)
• Timestamps - Audit trail (created_at, updated_at)
• Metadata - Flexible key-value storage for extensibility

What users-core does NOT do

• ❌ Authentication (passwords, tokens, sessions) → Use auth-core
• ❌ User profiles (names, addresses, bio, preferences) → Use profiles-core
• ❌ Permissions/Roles (RBAC, ACL) → Use permissions-core

Core Principle: users-core is the minimal registry of "who exists in the system". All other concerns are delegated to specialized packages through event-driven communication.

---

Features

• ✅ Framework-agnostic - Works with FastAPI, Flask, Django, CLI, desktop apps
• ✅ Database-agnostic - Bring your own repository (SQLAlchemy, MongoDB, or custom)
• ✅ Event-driven ready - Publishes domain events for inter-package communication
• ✅ Type-safe - Full type hints with Pydantic v2 validation
• ✅ Hexagonal architecture - Clean separation of domain, interfaces, and adapters
• ✅ Dependency injectable - All dependencies are interfaces, not implementations
• ✅ Migration support - Built-in migration providers for SQLAlchemy and MongoDB
• ✅ Fully tested - Comprehensive unit, integration, and contract tests

---

Installation

# Core package only (minimal dependencies)
pip install users-core

# With SQLAlchemy adapter
pip install users-core[sqlalchemy]

# With MongoDB adapter
pip install users-core[mongodb]

# With PostgreSQL
pip install users-core[postgres]

# With MySQL
pip install users-core[mysql]

# With Redis event bus
pip install users-core[events-redis]

# All adapters + dev tools
pip install users-core[all]

---

Quick Start

1. Basic Usage (In-Memory)

from users_core import UserService
from users_core.adapters.repositories import InMemoryUserRepository

# Setup
repository = InMemoryUserRepository()
service = UserService(repository=repository)

# Create user
user = service.create_user(
    email="john@example.com",
    username="johndoe",
    metadata={"source": "web", "referrer": "homepage"}
)

print(f"Created: {user.id}")  # UUID v4
print(f"Status: {user.status}")  # UserStatus.ACTIVE

# Get user
user = service.get_user(user.id)
user = service.get_user_by_email("john@example.com")
user = service.get_user_by_username("johndoe")

# Update
service.update_email(user.id, "newemail@example.com")
service.update_username(user.id, "john_doe")
service.update_metadata(user.id, {"premium": True})

# Status management
service.suspend_user(user.id, reason="Terms violation")
service.activate_user(user.id)
service.archive_user(user.id)

# Soft delete (status = DELETED, data retained)
service.delete_user(user.id, soft=True)

2. With SQLAlchemy (PostgreSQL/MySQL/SQLite)

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from users_core import UserService
from users_core.adapters.repositories import SQLAlchemyUserRepository

# Setup database
engine = create_engine("postgresql://user:pass@localhost/mydb")
Session = sessionmaker(bind=engine)
session = Session()

# Create repository
repository = SQLAlchemyUserRepository(session)
service = UserService(repository=repository)

# Use service (same API as in-memory)
user = service.create_user(
    email="alice@example.com",
    username="alice"
)

3. With MongoDB

from pymongo import MongoClient
from users_core import UserService
from users_core.adapters.repositories import MongoUserRepository

# Setup MongoDB
client = MongoClient("mongodb://localhost:27017")
database = client["myapp"]

# Create repository
repository = MongoUserRepository(database)
service = UserService(repository=repository)

# Use service (same API)
user = service.create_user(
    email="bob@example.com",
    username="bob"
)

4. With Event Bus (Event-Driven Architecture)

from users_core import UserService
from users_core.adapters.repositories import InMemoryUserRepository
from users_core.adapters.event_buses import InMemoryEventBus
from users_core.events import UserCreatedEvent, UserStatusChangedEvent

# Setup event bus
event_bus = InMemoryEventBus()

# Subscribe to events
def on_user_created(event: UserCreatedEvent):
    print(f"[EVENT] User created: {event.username} ({event.email})")
    # Trigger: Create auth credentials, initialize profile, send welcome email

def on_status_changed(event: UserStatusChangedEvent):
    print(f"[EVENT] Status changed: {event.old_status} -> {event.new_status}")
    # Trigger: Revoke sessions, notify admin, update analytics

event_bus.subscribe(UserCreatedEvent, on_user_created)
event_bus.subscribe(UserStatusChangedEvent, on_status_changed)

# Create service with event bus
repository = InMemoryUserRepository()
service = UserService(repository=repository, event_bus=event_bus)

# Events are published automatically
user = service.create_user(email="eve@example.com", username="eve")
# Output: [EVENT] User created: eve (eve@example.com)

service.suspend_user(user.id, reason="Fraud detection")
# Output: [EVENT] Status changed: active -> suspended

---

Database Migrations

users-core provides migration providers that implement the IMigrationProvider interface, allowing integration with global migration orchestrators.

SQLAlchemy Migrations

from sqlalchemy import create_engine
from users_core.adapters.migrations import SQLAlchemyMigrationProvider

engine = create_engine("postgresql://user:pass@localhost/mydb")
provider = SQLAlchemyMigrationProvider(engine)

# Check pending migrations
pending = provider.get_pending_migrations()
print(f"Pending: {[m.id for m in pending]}")

# Apply all pending migrations
applied = provider.apply_all_pending()
print(f"Applied: {applied}")

# Validate migration integrity (checksums)
provider.validate_migrations()

MongoDB Migrations

from pymongo import MongoClient
from users_core.adapters.migrations import MongoMigrationProvider

client = MongoClient("mongodb://localhost:27017")
database = client["myapp"]
provider = MongoMigrationProvider(database)

# Apply all pending migrations
applied = provider.apply_all_pending()
print(f"Applied: {applied}")

Global Migration Orchestration

Integrate with other packages' migrations:

from users_core.adapters.migrations import SQLAlchemyMigrationProvider
from profiles_core.adapters.migrations import ProfileMigrationProvider
from auth_core.adapters.migrations import AuthMigrationProvider

# Collect all migration providers
providers = [
    SQLAlchemyMigrationProvider(users_engine),
    ProfileMigrationProvider(profiles_engine),
    AuthMigrationProvider(auth_engine),
]

# Apply migrations across all packages
for provider in providers:
    print(f"Migrating: {provider.__class__.__name__}")
    applied = provider.apply_all_pending()
    provider.validate_migrations()

See migrations/README.md for details.

---

Architecture

users-core/
├── domain/                  # Core business logic (NO external dependencies)
│   ├── models.py            # User, UserStatus (dataclasses)
│   ├── services.py          # UserService (business logic)
│   └── exceptions.py        # Domain-specific exceptions
│
├── interfaces/              # Abstract contracts (Ports)
│   ├── repository.py        # IUserRepository (ABC)
│   ├── event_bus.py         # IEventBus (ABC)
│   └── migration.py         # IMigrationProvider (ABC)
│
├── dto/                     # Data Transfer Objects
│   ├── requests.py          # Input DTOs (with Pydantic validation)
│   └── responses.py         # Output DTOs (serialization)
│
├── events/                  # Domain events
│   └── events.py            # UserCreatedEvent, UserUpdatedEvent, etc.
│
├── adapters/                # Concrete implementations (OPTIONAL)
│   ├── repositories/        # InMemory, SQLAlchemy, MongoDB
│   ├── event_buses/         # InMemory, Redis
│   └── migrations/          # SQLAlchemy, MongoDB migration providers
│
├── utils/                   # Utilities
│   ├── validators.py        # Username/email validation
│   └── generators.py        # UUID generation
│
└── migrations/              # Migration scripts
    └── README.md            # Migration documentation

Dependency Flow

┌─────────────────────────────────────┐
│      Your Application Layer         │
│  (FastAPI, Flask, CLI, Desktop UI)  │
└──────────────┬──────────────────────┘
               │
        ┌──────▼──────┐
        │  Adapters   │ ← SQLAlchemy, MongoDB, Redis (OPTIONAL)
        └──────┬──────┘
               │
        ┌──────▼──────┐
        │   Service   │ ← UserService (business logic)
        └──────┬──────┘
               │
        ┌──────▼──────┐
        │   Domain    │ ← User, UserStatus (pure logic)
        └─────────────┘
               ▲
               │
        ┌──────┴──────┐
        │ Interfaces  │ ← IUserRepository, IEventBus (contracts)
        └─────────────┘

Golden Rule: Dependencies point INWARD. Domain knows nothing about adapters.

---

Domain Model

from enum import Enum
from dataclasses import dataclass
from datetime import datetime

class UserStatus(str, Enum):
    ACTIVE = "active"           # Normal, active user
    SUSPENDED = "suspended"     # Temporarily blocked
    ARCHIVED = "archived"       # Inactive, but data retained
    DELETED = "deleted"         # Soft-deleted

@dataclass
class User:
    id: str                     # UUID v4
    email: str                  # Unique, lowercase
    username: str               # Unique, lowercase, 3-30 chars
    status: UserStatus          # Current lifecycle state
    created_at: datetime        # UTC timestamp
    updated_at: datetime        # UTC timestamp
    metadata: dict              # Flexible key-value storage

Username Validation Rules

• Length: 3-30 characters
• Start: Must start with alphanumeric (a-z, 0-9)
• Characters: Letters, numbers, underscore (_), hyphen (-)
• Case: Case-insensitive (stored as lowercase)
• Uniqueness: Unique across the system

Valid: johndoe, user_123, alice-smith, bob42 Invalid: ab (too short), _john (invalid start), john@doe (invalid char), JOHNDOE (converted to johndoe)

Email Validation Rules

• Format: Standard email format (user@domain.com)
• Case: Case-insensitive (stored as lowercase)
• Uniqueness: Unique across the system

---

API Examples

FastAPI Integration

from fastapi import FastAPI, HTTPException, Depends
from users_core import UserService, UserNotFoundError, DuplicateEmailError
from users_core.dto import CreateUserRequest, UserResponse

app = FastAPI()

# Dependency injection
def get_user_service() -> UserService:
    repository = get_repository()  # Your implementation
    event_bus = get_event_bus()    # Optional
    return UserService(repository=repository, event_bus=event_bus)

@app.post("/users", response_model=UserResponse, status_code=201)
def create_user(
    request: CreateUserRequest,
    service: UserService = Depends(get_user_service)
):
    try:
        user = service.create_user(
            email=request.email,
            username=request.username,
            metadata=request.metadata or {}
        )
        return UserResponse.from_domain(user)
    except DuplicateEmailError:
        raise HTTPException(status_code=409, detail="Email already exists")

@app.get("/users/{user_id}", response_model=UserResponse)
def get_user(
    user_id: str,
    service: UserService = Depends(get_user_service)
):
    try:
        user = service.get_user(user_id)
        return UserResponse.from_domain(user)
    except UserNotFoundError:
        raise HTTPException(status_code=404, detail="User not found")

@app.patch("/users/{user_id}/email")
def update_email(
    user_id: str,
    email: str,
    service: UserService = Depends(get_user_service)
):
    try:
        service.update_email(user_id, email)
        return {"status": "ok"}
    except UserNotFoundError:
        raise HTTPException(status_code=404, detail="User not found")
    except DuplicateEmailError:
        raise HTTPException(status_code=409, detail="Email already exists")

@app.post("/users/{user_id}/suspend")
def suspend_user(
    user_id: str,
    reason: str,
    service: UserService = Depends(get_user_service)
):
    try:
        service.suspend_user(user_id, reason=reason)
        return {"status": "suspended"}
    except UserNotFoundError:
        raise HTTPException(status_code=404, detail="User not found")

Flask Integration

from flask import Flask, request, jsonify
from users_core import UserService, UserNotFoundError, DuplicateEmailError
from users_core.dto import UserResponse

app = Flask(__name__)

# Initialize service (you might use Flask extensions for this)
repository = get_repository()
user_service = UserService(repository=repository)

@app.route("/users", methods=["POST"])
def create_user():
    data = request.get_json()
    try:
        user = user_service.create_user(
            email=data["email"],
            username=data["username"],
            metadata=data.get("metadata", {})
        )
        return jsonify(UserResponse.from_domain(user).model_dump()), 201
    except DuplicateEmailError:
        return jsonify({"error": "Email already exists"}), 409

@app.route("/users/<user_id>", methods=["GET"])
def get_user(user_id):
    try:
        user = user_service.get_user(user_id)
        return jsonify(UserResponse.from_domain(user).model_dump())
    except UserNotFoundError:
        return jsonify({"error": "User not found"}), 404

---

Domain Events

When an event bus is provided, the service publishes these events:

Event	Trigger	Payload
UserCreatedEvent	User created	user_id, email, username, status, metadata
UserUpdatedEvent	Email/username/metadata updated	user_id, email, username, metadata
UserStatusChangedEvent	Status transition	user_id, old_status, new_status, reason
UserDeletedEvent	User deleted (soft/hard)	user_id, soft_delete

Event-Driven Use Cases

Other packages can subscribe to these events:

# auth-core: Create credentials when user is created
event_bus.subscribe(UserCreatedEvent, lambda e: auth_service.create_credentials(e.user_id))

# profiles-core: Initialize profile when user is created
event_bus.subscribe(UserCreatedEvent, lambda e: profile_service.create_profile(e.user_id))

# notifications-core: Send welcome email
event_bus.subscribe(UserCreatedEvent, lambda e: notification_service.send_welcome(e.email))

# auth-core: Revoke sessions when user is suspended
event_bus.subscribe(UserStatusChangedEvent, lambda e:
    auth_service.revoke_sessions(e.user_id) if e.new_status == "suspended" else None
)

# analytics-core: Track user lifecycle
event_bus.subscribe(UserStatusChangedEvent, lambda e: analytics_service.track(e))

---

Testing

Run Tests

# All tests
pytest

# With coverage report
pytest --cov

# Only unit tests (fast, no DB)
pytest tests/unit -v

# Only integration tests (with DB)
pytest tests/integration -v

# Parallel execution
pytest -n auto

Test Structure

tests/
├── unit/                    # Fast, isolated tests (domain logic)
│   ├── test_user_model.py
│   ├── test_user_service.py
│   └── test_validators.py
│
├── integration/             # Tests with real dependencies
│   ├── test_sqlalchemy_repository.py
│   ├── test_mongo_repository.py
│   └── test_migrations.py
│
├── contracts/               # Interface compliance tests
│   └── test_repository_contract.py
│
└── fixtures/
    └── conftest.py          # Shared test fixtures

Writing Tests

# Unit test (domain layer)
from users_core.domain import User, UserStatus

def test_user_creation():
    user = User.create(
        email="test@example.com",
        username="testuser"
    )
    assert user.status == UserStatus.ACTIVE
    assert user.email == "test@example.com"

# Integration test (with repository)
from users_core import UserService
from users_core.adapters.repositories import InMemoryUserRepository

def test_create_and_retrieve_user():
    repository = InMemoryUserRepository()
    service = UserService(repository=repository)

    user = service.create_user(
        email="test@example.com",
        username="testuser"
    )

    retrieved = service.get_user(user.id)
    assert retrieved.email == "test@example.com"

---

Development

Setup

# Clone repository
git clone https://github.com/Morgiver/users-core.git
cd users-core

# Install in development mode with all dependencies
pip install -e ".[all]"

# Run tests
pytest

# Check coverage
pytest --cov --cov-report=html
open htmlcov/index.html

Code Quality

# Format code
black src tests
isort src tests

# Lint
ruff check src tests

# Type check
mypy src

# Run all quality checks
black src tests && isort src tests && ruff check src tests && mypy src && pytest

Quality Metrics

• Coverage: >80% overall, >90% on domain layer
• Type safety: 100% type coverage (mypy strict mode)
• Linting: Zero ruff/black/isort violations

---

Design Principles

This package follows:

• Hexagonal Architecture (Ports & Adapters) - Domain isolated from infrastructure
• Domain-Driven Design (DDD) - Rich domain models, ubiquitous language
• SOLID Principles - Single responsibility, dependency inversion
• Dependency Injection - All dependencies are interfaces
• Event-Driven Architecture - Loose coupling via domain events
• Single Responsibility - Only manages user identity, nothing else

---

Related Packages

Build a complete identity system with these complementary packages:

• auth-core - Authentication (passwords, tokens, sessions, OAuth)
• profiles-core - Flexible user profiles (names, addresses, preferences)
• permissions-core - Role-based access control (RBAC, ACL)
• notifications-core - Email, SMS, push notifications
• audit-core - Activity logging and compliance

---

License

MIT License - see LICENSE file for details.

---

Contributing

Contributions welcome! Please:

1. Fork the repository
2. Create a feature branch (git checkout -b feature/amazing-feature)
3. Write tests for new functionality
4. Ensure all tests pass (pytest)
5. Follow code quality standards (black, ruff, mypy)
6. Submit a pull request

---

Support

• Documentation: This README + inline docstrings
• Issues: GitHub Issues
• Discussions: GitHub Discussions

---

Built with care following Hexagonal Architecture and Domain-Driven Design principles.