diff --git a/.env b/.env deleted file mode 100644 index fb7668b..0000000 --- a/.env +++ /dev/null @@ -1,3 +0,0 @@ -ADMIN_USER=admin -ADMIN_PASS=pass -ADMIN_SECRET=sec diff --git a/.env.example b/.env.example index 2c3f018..a18634c 100644 --- a/.env.example +++ b/.env.example @@ -1 +1,8 @@ -ADMIN_USER=admin ADMIN_PASS=pass ADMIN_SECRET=sec +# Database connection string used by the generated application +DATABASE_URL=sqlite+aiosqlite:///./database.db + +# SQLAdmin authentication (only required when generating with --admin-auth) +ADMIN_USER=admin +ADMIN_PASS=change-me +# ADMIN_SECRET must be at least 32 characters long +ADMIN_SECRET=replace-with-a-secret-of-at-least-32-characters diff --git a/.github/workflows/publish.yml b/.github/workflows/publish.yml new file mode 100644 index 0000000..472b0d2 --- /dev/null +++ b/.github/workflows/publish.yml @@ -0,0 +1,59 @@ +name: Publish to PyPI + +# Publishes the package to PyPI when a GitHub Release is published. +# Uses PyPI Trusted Publishing (OIDC) — no API tokens are stored in the repo. +# +# One-time setup on PyPI (https://pypi.org/manage/account/publishing/): +# - Project name: dbml-to-sqlmodel +# - Owner: azhig +# - Repository: dbml-to-sqlmodel +# - Workflow name: publish.yml +# - Environment: pypi + +on: + release: + types: [published] + workflow_dispatch: + +jobs: + build: + name: Build distributions + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v6 + + - name: Install uv + uses: astral-sh/setup-uv@v7 + with: + version: "latest" + + - name: Build sdist and wheel + run: uv build + + - name: Check distribution metadata + run: uvx twine check dist/* + + - name: Upload distributions + uses: actions/upload-artifact@v4 + with: + name: dist + path: dist/ + + publish: + name: Publish to PyPI + needs: build + runs-on: ubuntu-latest + environment: + name: pypi + url: https://pypi.org/project/dbml-to-sqlmodel/ + permissions: + id-token: write # required for trusted publishing + steps: + - name: Download distributions + uses: actions/download-artifact@v4 + with: + name: dist + path: dist/ + + - name: Publish to PyPI + uses: pypa/gh-action-pypi-publish@release/v1 diff --git a/.gitignore b/.gitignore index 7275c9d..cd0a45a 100644 --- a/.gitignore +++ b/.gitignore @@ -19,7 +19,7 @@ habr_article.md *.db-shm *.db-wal -# Generated files in root (legacy - если остались) +# Generated files in root /models.py /admin.py /crud/ @@ -35,6 +35,7 @@ habr_article.md .pytest_cache/ .ruff_cache/ .coverage +.mypy_cache htmlcov/ # OS diff --git a/CHANGELOG.md b/CHANGELOG.md index 83d3563..40b3ec8 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,41 +5,39 @@ All notable changes to this project will be documented in this file. The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/), and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). -## [Unreleased] +## [0.1.0] - 2026-06-02 -### Added -- Ruff configuration for linting and formatting -- Pre-commit hooks for code quality -- MyPy type checking support -- Dependabot for automated dependency updates -- CHANGELOG.md for tracking version changes -- CONTRIBUTING.md guidelines - -### Changed -- Project structure refactored from `dbml_to_code` to `dbml_to_sqlmodel` -- Updated Makefile with correct CLI commands -- Improved .gitignore patterns - -## [0.1.0] - 2026-01-14 +Initial public release. ### Added -- Initial release -- DBML to SQLModel conversion -- FastAPI CRUD generation -- Interactive CLI with questionary -- Preview mode with diff display -- Info command for file inspection -- GitHub Actions CI/CD pipeline -- Test coverage reporting -- Example schema files - -### Features -- Generate SQLModel models from DBML schemas -- Auto-generate FastAPI CRUD endpoints -- SQLAdmin integration -- Database migrations support -- Type-safe code generation -- CLI with multiple commands (generate, preview, info) -[Unreleased]: https://github.com/yourusername/dbml-to-sqlmodel/compare/v0.1.0...HEAD -[0.1.0]: https://github.com/yourusername/dbml-to-sqlmodel/releases/tag/v0.1.0 +- DBML to SQLModel model generation +- FastAPI CRUD endpoint generation powered by + [FastCRUD](https://github.com/igorbenav/fastcrud) +- Optional [SQLAdmin](https://github.com/aminalaee/sqladmin) admin panel, with + optional constant-time credential authentication (`--admin-auth`) +- Modular generated project: one package per table, plus a wired-up `main.py` + that uses the FastAPI `lifespan` API and reads `DATABASE_URL` from the + environment (`.env`) +- Correct temporal type mapping: DBML `timestamp`/`datetime`/`date`/`time` map to + Python `datetime`/`date`/`time` with the matching imports +- PascalCase class names and singularized relationship attribute names +- Interactive CLI (`questionary`) plus direct subcommands: `generate`, + `preview`, `info` and `code-to-dbml` +- Preview mode with diff display and protection for manual edits via the + `USER_MODIFIED` marker +- Reverse conversion: generated code back to DBML +- `--version` / `-V` flag +- Generated `requirements.txt` with runtime versions aligned to the tested + stack (including `greenlet`, which the async SQLAlchemy engine needs but + which is not auto-installed on every platform) +- `py.typed` marker (PEP 561) and full PyPI packaging metadata (authors, + keywords, Trove classifiers, project URLs) +- Documentation: README, CLI guide and contributing guidelines +- Tooling: Ruff, MyPy, pre-commit, Dependabot, GitHub Actions CI with Codecov + coverage and a PyPI trusted-publishing workflow +- Test suite with 100% coverage, including smoke tests that compile the + generated application and guard temporal-type, lifespan and requirements + regressions + +[0.1.0]: https://github.com/azhig/dbml-to-sqlmodel/releases/tag/v0.1.0 diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index 8b75d46..d71153f 100644 --- a/CONTRIBUTING.md +++ b/CONTRIBUTING.md @@ -6,8 +6,8 @@ Thank you for your interest in contributing! This document provides guidelines f 1. **Clone the repository** ```bash - git clone - cd dbml_to_crud + git clone https://github.com/azhig/dbml-to-sqlmodel.git + cd dbml-to-sqlmodel ``` 2. **Install uv** (if not already installed) @@ -232,18 +232,22 @@ docs(readme): update installation instructions ## Project Structure ``` -dbml_to_crud/ +dbml-to-sqlmodel/ ├── src/ │ └── dbml_to_sqlmodel/ # Main package │ ├── cli.py # CLI entry point -│ ├── parser.py # DBML parser │ ├── generator.py # Code generator -│ ├── config.py # Configuration -│ └── templates/ # Jinja2 templates +│ ├── core/ # Parser, config, code generation +│ ├── commands/ # CLI subcommands (generate, preview, info, code-to-dbml) +│ ├── integrations/ # pydbml adapter +│ ├── models/ # Internal data models +│ ├── utils/ # Helpers (diff, formatters, file manager) +│ └── templates/ # Code templates ├── tests/ # Test files ├── examples/ # Example schemas +├── docs/ # Documentation ├── Makefile # Development commands -└── pyproject.toml # Project metadata +└── pyproject.toml # Project metadata ``` ## Configuration Files diff --git a/Makefile b/Makefile index abce17d..9c7ceeb 100644 --- a/Makefile +++ b/Makefile @@ -1,4 +1,4 @@ -.PHONY: help install cli generate preview info run dev clean db-reset format format-imports lint lint-fix typecheck check pre-commit pre-commit-install fresh restart full-reset +.PHONY: help install cli generate preview info run-deps run dev clean db-reset format format-imports lint lint-fix typecheck check pre-commit pre-commit-install fresh restart full-reset # Output colors GREEN := \033[0;32m @@ -27,10 +27,14 @@ preview: ## Show diff without writing files info: ## Show generated files and mismatches uv run dbml-to-sqlmodel info examples/schema.dbml -run: ## Run server (production) +run-deps: ## Install runtime dependencies for the generated app (fastapi, sqlmodel, greenlet, ...) + uv sync --extra runtime + @echo "$(GREEN)OK: runtime dependencies installed$(NC)" + +run: run-deps ## Run server (production) cd output && uv run python main.py -dev: ## Run server in development mode (hot reload) +dev: run-deps ## Run server in development mode (hot reload) cd output && uv run uvicorn main:app --reload --host 0.0.0.0 --port 8001 clean: ## Remove generated files and caches diff --git a/README.md b/README.md index 01d5de6..ad46985 100644 --- a/README.md +++ b/README.md @@ -15,15 +15,20 @@ ```

- - Tests + + PyPI version - - Coverage + + Python Versions - Python Versions - - License: MIT + + Tests + + + Coverage + + + License: MIT Ruff @@ -31,50 +36,57 @@ Pre-commit - Code style: ruff Type checked: mypy

-**Generate FastAPI + SQLModel + FastCRUD + SQLAdmin from DBML schema** +**Generate FastAPI + SQLModel + FastCRUD + SQLAdmin applications from a DBML schema.** + +`dbml-to-sqlmodel` turns a [DBML](https://dbml.org/) database schema into a ready-to-run, +modular FastAPI project: SQLModel models, FastCRUD routers, an optional SQLAdmin panel, +and a wired-up application — in a single command. ## Features -- Generate SQLModel models from DBML schema -- Auto-create CRUD routers with FastCRUD -- Generate FastAPI application with ready-to-use endpoints -- Optional SQLAdmin panel -- Preview mode to review changes before applying -- Protect user modifications in generated files +- Generate SQLModel models from a DBML schema +- Auto-generate CRUD routers powered by [FastCRUD](https://github.com/igorbenav/fastcrud) +- Produce a FastAPI application with ready-to-use endpoints +- Optional [SQLAdmin](https://github.com/aminalaee/sqladmin) admin panel +- Preview mode to inspect changes before applying them +- Protection for your manual edits in generated files (`USER_MODIFIED` marker) +- Reverse conversion: generated code back to DBML - Interactive CLI and direct commands ## Requirements - Python 3.11+ -- uv (recommended) or any Python environment manager +- [uv](https://docs.astral.sh/uv/) (recommended) or any other Python environment manager ## Installation -### For Code Generation Only +### Code generation only ```bash pip install dbml-to-sqlmodel ``` -### For Running Generated Applications +### To run the generated applications ```bash # Install with runtime dependencies -pip install dbml-to-sqlmodel[runtime] +pip install "dbml-to-sqlmodel[runtime]" + +# Or install the runtime dependencies separately in your project +pip install "fastapi[all]" sqlmodel fastcrud sqladmin aiosqlite -# Or install runtime dependencies separately in your project -pip install fastapi[all] sqlmodel fastcrud sqladmin aiosqlite +# Or, when working inside this repository +uv sync --extra runtime ``` ## Quick Start ### 1. Create a DBML schema -Create a `schema.dbml` file with your database structure: +Create a `schema.dbml` file describing your database. For example: ```dbml Table users { @@ -93,38 +105,63 @@ Table posts { } ``` +![DBML schema](https://raw.githubusercontent.com/azhig/dbml-to-sqlmodel/main/docs/img/dbml.png) + ### 2. Generate the application ```bash dbml-to-sqlmodel generate schema.dbml -o output ``` -### 3. Setup environment +Or launch the interactive mode: + +```bash +dbml-to-sqlmodel +# or the alternative command +dbml2sm +``` + +![Interactive CLI](https://raw.githubusercontent.com/azhig/dbml-to-sqlmodel/main/docs/img/cli.png) + +After generation, all SQLModel objects and their `FastCRUD` routers are created +automatically in the target directory. + +### 3. Configure the environment ```bash cd output echo "DATABASE_URL=sqlite+aiosqlite:///./database.db" > .env -# Install runtime dependencies if not already installed -pip install dbml-to-sqlmodel[runtime] +# Install the runtime dependencies if they are not installed yet +pip install "dbml-to-sqlmodel[runtime]" ``` ### 4. Run the application ```bash python main.py + +# From this repository you can also run it via Make. +# `make run` first installs the runtime dependencies (uv sync --extra runtime) +# and then starts the server; `make dev` does the same with hot reload. +make run ``` -### 5. Open in browser +### 5. Open it in the browser - API documentation: `http://localhost:8001/docs` -- Admin panel: `http://localhost:8001/admin` + +![API docs](https://raw.githubusercontent.com/azhig/dbml-to-sqlmodel/main/docs/img/docs.png) + +- SQLAdmin panel: `http://localhost:8001/admin` + +![Admin panel](https://raw.githubusercontent.com/azhig/dbml-to-sqlmodel/main/docs/img/admin.png) ## Generated Structure -The generator creates a modular project where each table has its own directory: +The generator produces a modular project with a dedicated directory per table: -``` +```text output/ ├── main.py # FastAPI application ├── admin.py # SQLAdmin configuration @@ -142,36 +179,97 @@ output/ └── ... (one directory per table) ``` -For detailed structure documentation, see [CLI_GUIDE.md](CLI_GUIDE.md#generated-project-structure). +For a detailed description of the structure, see the +[CLI Guide](https://github.com/azhig/dbml-to-sqlmodel/blob/main/docs/CLI_GUIDE.md). + +## Using It in Your Own Project + +`dbml-to-sqlmodel` is a **code generator**: you need it while developing (to scaffold +and regenerate code from your schema), but the running application never imports it. +The recommended setup is therefore: + +- add the generator itself as a **development** dependency, and +- add the libraries the generated app uses as your project's **runtime** dependencies. + +### 1. Add the generator as a dev dependency + +```bash +# uv (recommended) +uv add --dev dbml-to-sqlmodel + +# Poetry +poetry add --group dev dbml-to-sqlmodel + +# pip (e.g. into requirements-dev.txt) +pip install dbml-to-sqlmodel +``` + +### 2. Add the runtime libraries the generated app needs + +The generated project imports FastAPI, SQLModel, FastCRUD, SQLAdmin and an async +database driver. Add them as regular (runtime) dependencies of your project: + +```bash +uv add "fastapi[all]" sqlmodel fastcrud sqladmin aiosqlite greenlet +``` + +> `greenlet` is required by SQLAlchemy's async engine and is **not** auto-installed +> on every platform/Python combination, so add it explicitly — otherwise the app +> fails to start with `the greenlet library is required`. + +### 3. Recommended workflow + +1. Keep `schema.dbml` under version control and treat it as the single source of truth. +2. Regenerate the code whenever the schema changes: + ```bash + uv run dbml-to-sqlmodel generate schema.dbml -o output + ``` +3. Preview the changes before applying them to an existing project: + ```bash + uv run dbml-to-sqlmodel preview schema.dbml -o output + ``` +4. Protect any file you edit by hand with a `# USER_MODIFIED` marker (see + [Protecting Your Modifications](#protecting-your-modifications)); such files are + kept on regeneration unless you pass `--force`. +5. If you adjusted the models by hand, you can sync the schema back from the code: + ```bash + uv run dbml-to-sqlmodel code-to-dbml output -o schema.dbml + ``` +6. Commit both `schema.dbml` and the generated code so the repository stays reproducible. ## CLI Usage -For complete CLI documentation, see [CLI_GUIDE.md](CLI_GUIDE.md). +Full CLI reference: [CLI Guide](https://github.com/azhig/dbml-to-sqlmodel/blob/main/docs/CLI_GUIDE.md). ### Quick Commands +Everywhere `dbml-to-sqlmodel` can be replaced with the shorter `dbml2sm`. + ```bash # Interactive CLI mode dbml-to-sqlmodel -# Generate application +# Generate the application dbml-to-sqlmodel generate schema.dbml -o output # Preview changes dbml-to-sqlmodel preview schema.dbml -# Show schema info +# Schema information dbml-to-sqlmodel info schema.dbml -# Reverse: code to DBML +# Reverse conversion: code -> DBML dbml-to-sqlmodel code-to-dbml output -o schema.dbml + +# Show the installed version +dbml-to-sqlmodel --version ``` ## Configuration ### Environment Variables -Create a `.env` file in your project directory: +Create a `.env` file in the project root: ```env DATABASE_URL=sqlite+aiosqlite:///./database.db @@ -179,13 +277,13 @@ DATABASE_URL=sqlite+aiosqlite:///./database.db ### SQLAdmin Authentication -To enable admin panel authentication: +To enable authentication for the admin panel: ```bash dbml-to-sqlmodel generate schema.dbml --admin-auth ``` -Add to your `.env` file: +Add the following to `.env`: ```env ADMIN_USER=admin @@ -193,47 +291,21 @@ ADMIN_PASS=your-secure-password ADMIN_SECRET=your-secret-key-must-be-at-least-32-characters-long ``` -For more configuration options, see [CLI_GUIDE.md](CLI_GUIDE.md). +![Admin login](https://raw.githubusercontent.com/azhig/dbml-to-sqlmodel/main/docs/img/admin_login.png) + +Other configuration options are described in the +[CLI Guide](https://github.com/azhig/dbml-to-sqlmodel/blob/main/docs/CLI_GUIDE.md). ## Protecting Your Modifications -If you modify generated files and want to prevent them from being overwritten on regeneration, add this comment at the top of the file: +If you edit generated files and want to protect them from being overwritten on +regeneration, add the following comment at the top of the file: ```python # USER_MODIFIED ``` -Files with this marker will be protected during regeneration unless you use the `--force` flag. - -## Development - -### Running Tests - -```bash -# Run all tests -make test - -# Run tests with coverage -make coverage - -# Generate HTML coverage report -make coverage-html -# Open htmlcov/index.html in browser -``` - -### Code Quality - -```bash -# Format code -make format - -# Lint code -make lint -``` - -## Usage Examples - -See [CLI_GUIDE.md](CLI_GUIDE.md) for complete examples and workflows. +Files with this marker are not overwritten unless you pass the `--force` flag. ## Table Relationships @@ -252,6 +324,19 @@ Table posts { } ``` +## Documentation + +- [CLI Guide](https://github.com/azhig/dbml-to-sqlmodel/blob/main/docs/CLI_GUIDE.md) — full command reference and usage examples +- [Changelog](https://github.com/azhig/dbml-to-sqlmodel/blob/main/CHANGELOG.md) +- [Contributing](https://github.com/azhig/dbml-to-sqlmodel/blob/main/CONTRIBUTING.md) + +## Contributing + +Contributions are welcome! Please read the +[contributing guidelines](https://github.com/azhig/dbml-to-sqlmodel/blob/main/CONTRIBUTING.md) +before opening a pull request. + ## License -MIT +This project is licensed under the terms of the +[MIT License](https://github.com/azhig/dbml-to-sqlmodel/blob/main/LICENSE). diff --git a/docs/CLI_GUIDE.md b/docs/CLI_GUIDE.md index d7faf7c..2807529 100644 --- a/docs/CLI_GUIDE.md +++ b/docs/CLI_GUIDE.md @@ -1,6 +1,6 @@ -# CLI Usage Guide +# CLI Guide -Complete guide for using the `dbml-to-sqlmodel` command-line interface. +Complete guide to using the `dbml-to-sqlmodel` CLI. ## Table of Contents @@ -14,724 +14,194 @@ Complete guide for using the `dbml-to-sqlmodel` command-line interface. - [Configuration](#configuration) - [File Protection](#file-protection) - [Examples](#examples) +- [Troubleshooting](#troubleshooting) ## Installation -### For Code Generation Only - -Install just the generator tool (lightweight, no runtime dependencies): +### Code generation only ```bash pip install dbml-to-sqlmodel ``` -### For Running Generated Applications - -If you want to run the generated application, install with runtime dependencies: +### To run the generated applications ```bash -pip install dbml-to-sqlmodel[runtime] +pip install "dbml-to-sqlmodel[runtime]" ``` -Or install runtime dependencies separately in your project: +## Interactive Mode + +![Interactive mode — initial setup](img/cli_interactive_initial.png) ```bash -pip install fastapi[all] sqlmodel fastcrud sqladmin aiosqlite +dbml-to-sqlmodel +# or from the repository +make cli ``` -## Interactive Mode +On the very first run, an **Initial Setup** wizard lets you configure a few options. -Launch the interactive CLI menu: +**In order:** -```bash -dbml-to-sqlmodel +```text +1. Path to DBML schema: # Choose the DBML file +2. Output directory: # Output folder +3. Show all files in preview?: # Show the full list or only changes +4. Show new file contents?: # Preview the content of new files +5. Overwrite protected files (USER_MODIFIED)?: # Force-overwrite protected files +6. Require login for SQLAdmin?: # Whether to require an admin password ``` -Interactive mode provides a guided experience: +> The SQLAdmin credentials are managed through environment variables — see +> [.env.example](../.env.example). -1. **Select DBML file** - Browse and select your schema file -2. **Choose output directory** - Specify where to generate files -3. **Configure options** - Set generation parameters -4. **Preview mode** - Review changes before applying -5. **Save settings** - Store configuration for future use +After that, the interactive menu becomes available: -The interactive mode saves your preferences to `.dbml_to_code` in the project directory. +```text +1) [>] Settings # - Configure generator options +2) [>] DBML → Code # - Generate FastAPI from schema +3) [>] Code → DBML # - Extract schema from models +4) [>] Report # - View generation statistics +5) [x] Exit +``` ## Commands ### generate -Generate a complete FastAPI application from DBML schema. - -**Syntax:** +Generate a FastAPI application from DBML. ```bash -dbml-to-sqlmodel generate [OPTIONS] +dbml-to-sqlmodel generate [-o ] [--admin-auth] [--force] ``` -**Arguments:** - -- `schema_file` - Path to DBML schema file (required) - -**Options:** - -- `-o, --output ` - Output directory (default: `output`) -- `--admin-auth` - Enable authentication in SQLAdmin panel -- `--force` - Overwrite files marked as USER_MODIFIED -- `--help` - Show help message - -**Examples:** +**Example:** ```bash -# Basic generation -dbml-to-sqlmodel generate examples/schema.dbml - -# Custom output directory -dbml-to-sqlmodel generate examples/schema.dbml -o my_app - -# With admin authentication -dbml-to-sqlmodel generate examples/schema.dbml --admin-auth - -# Force overwrite protected files -dbml-to-sqlmodel generate examples/schema.dbml --force +dbml-to-sqlmodel generate schema.dbml -o my_app --admin-auth ``` -**Generated Structure:** - -``` -output/ -├── main.py # FastAPI application -├── admin.py # SQLAdmin configuration -├── requirements.txt # Project dependencies -└── models/ - ├── __init__.py # Root init (imports all models) - ├── enums.py # (Optional) DBML enums - ├── users/ - │ ├── __init__.py # Local init (exports models and router) - │ ├── model.py # SQLModel classes (Base, Main, Create, Update) - │ └── crud.py # FastCRUD router - ├── posts/ - │ ├── __init__.py - │ ├── model.py - │ └── crud.py - └── ... (one directory per table) -``` +![generate command](img/cli_generate.png) ### preview -Preview changes without applying them. Shows which files will be created, modified, or skipped. - -**Syntax:** +Preview changes **without writing** any files. ```bash -dbml-to-sqlmodel preview [OPTIONS] +dbml-to-sqlmodel preview schema.dbml -o my_app ``` -**Arguments:** - -- `schema_file` - Path to DBML schema file (required) - -**Options:** - -- `-o, --output ` - Output directory (default: `output`) -- `--help` - Show help message +![preview command](img/cli_preview.png) **Output:** -The preview shows a table with file statuses: - -- `CREATE` - New file will be created -- `UPDATE` - Existing file will be modified -- `SKIP` - File is protected (USER_MODIFIED marker) -- `UNCHANGED` - File content is identical - -**Examples:** - -```bash -# Preview changes -dbml-to-sqlmodel preview examples/schema.dbml - -# Preview with custom output directory -dbml-to-sqlmodel preview examples/schema.dbml -o my_app - -# Preview, then generate -dbml-to-sqlmodel preview examples/schema.dbml && dbml-to-sqlmodel generate schema.dbml +```text +📁 models/users/model.py UPDATE +📁 models/posts/crud.py CREATE +📁 users/__init__.py SKIP (USER_MODIFIED) ← Protected! ``` ### info -Display schema information: tables, columns, types, and relationships. - -**Syntax:** - -```bash -dbml-to-sqlmodel info -``` - -**Arguments:** - -- `schema_file` - Path to DBML schema file (required) - -**Output:** - -Shows: -- Table names and column counts -- Column details (name, type, constraints) -- Relationships between tables -- Primary keys and foreign keys - -**Example:** +Show information about the schema and the generated files. ```bash -dbml-to-sqlmodel info examples/schema.dbml +dbml-to-sqlmodel info schema.dbml ``` -**Sample Output:** - -``` -Schema: schema.dbml - -Table: users (4 columns) -├── id: int [PK] -├── username: str [unique, not null] -├── email: str [unique, not null] -└── created_at: datetime - -Table: posts (5 columns) -├── id: int [PK] -├── title: str [not null] -├── content: str -├── user_id: int [FK -> users.id] -└── created_at: datetime - -Relationships: - posts.user_id -> users.id (Many-to-One) -``` +![info command](img/cli_info.png) ### code-to-dbml -Reverse conversion: generate DBML schema from existing generated code. - -**Syntax:** +Reverse conversion: generated code → DBML. ```bash -dbml-to-sqlmodel code-to-dbml [OPTIONS] -``` - -**Arguments:** - -- `source_dir` - Directory with generated code (required) - -**Options:** - -- `-o, --output ` - Output DBML file (default: `schema.dbml`) -- `--help` - Show help message - -**Examples:** - -```bash -# Generate DBML from code -dbml-to-sqlmodel code-to-dbml output - -# Custom output file -dbml-to-sqlmodel code-to-dbml output -o new_schema.dbml - -# Update existing schema -dbml-to-sqlmodel code-to-dbml my_app -o examples/schema.dbml -``` - -**Use Cases:** - -1. Document existing generated code -2. Sync DBML after manual code changes -3. Create backup of current schema state -4. Compare schemas across versions - -## Generated Project Structure - -### Overview - -The generator creates a modular structure where each table gets its own directory with model and CRUD files: - -``` -output/ -├── main.py # FastAPI application entry point -├── admin.py # SQLAdmin panel configuration -├── requirements.txt # Project dependencies -├── .env # Environment variables (created manually) -└── models/ - ├── __init__.py # Exports all models and routers - ├── enums.py # (Optional) DBML enum definitions - │ - ├── users/ # Table: users - │ ├── __init__.py # Exports Users, UsersCreate, UsersUpdate, create_users_router - │ ├── model.py # SQLModel classes - │ └── crud.py # FastCRUD router factory - │ - ├── posts/ # Table: posts - │ ├── __init__.py - │ ├── model.py - │ └── crud.py - │ - └── categories/ # Table: categories - ├── __init__.py - ├── model.py - └── crud.py -``` - -### File Details - -#### `models/{table}/model.py` - -Contains 4 SQLModel classes for each table: - -```python -# Example: models/users/model.py - -from sqlmodel import SQLModel, Field -from typing import Optional - -# 1. Base class - shared fields (no PK) -class UsersBase(SQLModel): - username: str = Field(unique=True) - email: str = Field(unique=True) - full_name: Optional[str] = None - -# 2. Main table class - with PK and relationships -class Users(UsersBase, table=True): - """Users table""" - id: Optional[int] = Field(default=None, primary_key=True) - -# 3. Create schema - for POST requests (no PK) -class UsersCreate(UsersBase): - """Create schema for users""" - pass - -# 4. Update schema - for PATCH requests (all Optional) -class UsersUpdate(SQLModel): - """Update schema for users""" - username: Optional[str] = None - email: Optional[str] = None - full_name: Optional[str] = None -``` - -#### `models/{table}/crud.py` - -Contains router factory function: - -```python -# Example: models/users/crud.py - -from fastcrud import crud_router -from .model import Users, UsersCreate, UsersUpdate - -def create_users_router(get_session): - """Create CRUD router with session dependency""" - return crud_router( - model=Users, - create_schema=UsersCreate, - update_schema=UsersUpdate, - path='/users', - tags=['users'], - session=get_session - ) -``` - -#### `models/{table}/__init__.py` - -Exports classes and router: - -```python -# Example: models/users/__init__.py - -from .model import Users, UsersCreate, UsersUpdate -from .crud import create_users_router - -__all__ = [ - "Users", - "UsersCreate", - "UsersUpdate", - "create_users_router", -] -``` - -#### `models/__init__.py` - -Root init that exports everything: - -```python -# models/__init__.py - -from .users import Users, UsersCreate, UsersUpdate, create_users_router -from .posts import Posts, PostsCreate, PostsUpdate, create_posts_router -from .categories import Categories, CategoriesCreate, CategoriesUpdate, create_categories_router - -__all__ = [ - # Users - "Users", "UsersCreate", "UsersUpdate", "create_users_router", - # Posts - "Posts", "PostsCreate", "PostsUpdate", "create_posts_router", - # Categories - "Categories", "CategoriesCreate", "CategoriesUpdate", "create_categories_router", -] -``` - -#### `models/enums.py` - -Generated only if DBML contains enum definitions: - -```python -# models/enums.py - -from enum import Enum - -class StatusEnum(str, Enum): - """DBML enum: status""" - __dbml_name__ = "status" - ACTIVE = "active" - INACTIVE = "inactive" - PENDING = "pending" -``` - -#### `main.py` - -FastAPI application entry point: - -```python -# main.py (simplified) - -from fastapi import FastAPI -from sqlalchemy.ext.asyncio import create_async_engine -from sqlmodel import SQLModel - -# Import router factories -from models.users import create_users_router -from models.posts import create_posts_router - -app = FastAPI(title="Generated API") - -# Database setup -engine = create_async_engine("sqlite+aiosqlite:///./database.db") - -# Create session dependency -async def get_session(): - async with AsyncSession(engine) as session: - yield session - -# Initialize routers -users_router = create_users_router(get_session) -posts_router = create_posts_router(get_session) - -# Register routers -app.include_router(users_router) -app.include_router(posts_router) - -# Initialize admin panel -from admin import init_admin -init_admin(app, engine) -``` - -#### `admin.py` - -SQLAdmin panel configuration: - -```python -# admin.py (simplified) - -from sqladmin import Admin, ModelView -from models.users import Users -from models.posts import Posts - -def init_admin(app, engine): - admin = Admin(app, engine) - - class UsersAdmin(ModelView, model=Users): - name = "User" - name_plural = "Users" - icon = "fa-solid fa-user" - - class PostsAdmin(ModelView, model=Posts): - name = "Post" - name_plural = "Posts" - icon = "fa-solid fa-file-text" - - admin.add_view(UsersAdmin) - admin.add_view(PostsAdmin) - - return admin -``` - -#### `requirements.txt` - -Project dependencies: - -```txt -fastapi>=0.104.0 -sqlmodel>=0.0.14 -sqladmin>=0.22.0 -fastcrud>=0.20.1 -uvicorn[standard]>=0.23.0 -aiosqlite>=0.19.0 -python-dotenv>=1.0.0 -``` - -### Import Patterns - -**Import a model:** -```python -from models.users import Users, UsersCreate, UsersUpdate -``` - -**Import multiple models:** -```python -from models.users import Users -from models.posts import Posts -from models.categories import Categories +dbml-to-sqlmodel code-to-dbml output -o backup.dbml ``` -**Import from root (if needed):** -```python -from models import Users, Posts, Categories -``` - -### Benefits of This Structure - -1. **Modularity** - Each table is self-contained in its directory -2. **Scalability** - Easy to find files even with 50+ tables -3. **Extensibility** - Add custom files to table directories: - ``` - users/ - ├── model.py - ├── crud.py - ├── schemas.py # Custom Pydantic schemas - ├── services.py # Business logic - └── validators.py # Custom validators - ``` -4. **Isolation** - Changes to one table don't affect others -5. **Microservices Ready** - Easy to extract a table module into separate service +![code-to-dbml command](img/cli_code2dbml.png) ## Configuration -### Settings File - -CLI saves configuration to `.dbml_to_code` in your project directory. - -**Format:** +**.dbml_to_sqlmodel** (JSON, saved automatically): ```json -{ - "schema_file": "schema.dbml", - "output_dir": "output", - "preview_enabled": true, - "overwrite": false, - "admin_auth": false -} +{ "schema_file": "schema.dbml", "output_dir": "output" } ``` -**Parameters:** - -- `schema_file` - Default DBML file path -- `output_dir` - Default output directory -- `preview_enabled` - Show preview before generation -- `overwrite` - Overwrite USER_MODIFIED files -- `admin_auth` - Enable admin authentication - -### Environment Variables - -For generated applications, create `.env` file: - -**Basic Configuration:** +**.env** (for the generated applications): ```env -DATABASE_URL=sqlite+aiosqlite:///./database.db -``` - -**With Admin Authentication:** - -```env -DATABASE_URL=sqlite+aiosqlite:///./database.db +DATABASE_URL=sqlite+aiosqlite:///./db.db ADMIN_USER=admin -ADMIN_PASS=your-secure-password -ADMIN_SECRET=your-secret-key-must-be-at-least-32-characters-long +ADMIN_PASS=pass +ADMIN_SECRET=at_least_32_characters_secret ``` -**Security Notes:** - -- Never commit `.env` to version control -- Use strong passwords in production -- Keep `ADMIN_SECRET` at least 32 characters -- Rotate secrets regularly - ## File Protection -### Protecting Your Modifications - -Add a marker to prevent file overwriting: +Add this marker to the top of a file: ```python # USER_MODIFIED - -# Your custom code here -from sqlmodel import Field, SQLModel - -class User(SQLModel, table=True): - id: int | None = Field(default=None, primary_key=True) - username: str = Field(unique=True) - # Your custom fields... -``` - -**Behavior:** - -- Files with `# USER_MODIFIED` are skipped during regeneration -- Use `--force` flag to override protection -- Preview mode shows protected files as `SKIP` - -### Force Overwrite - -Override file protection: - -```bash -dbml-to-sqlmodel generate schema.dbml --force ``` -**Warning:** This will overwrite ALL files, including USER_MODIFIED ones. +**Result:** the file is shown as `SKIP` in preview; use `--force` to overwrite it. ## Examples -### Complete Workflow +### Full workflow ```bash -# 1. Create schema -cat > schema.dbml < schema.dbml <<'EOF' +Table users { id integer [pk], username varchar [unique] } EOF -# 2. Inspect schema -dbml-to-sqlmodel info schema.dbml - -# 3. Preview generation -dbml-to-sqlmodel preview schema.dbml +dbml-to-sqlmodel info schema.dbml # Inspect +dbml-to-sqlmodel preview schema.dbml # Preview +dbml-to-sqlmodel generate schema.dbml # Generate -# 4. Generate application -dbml-to-sqlmodel generate schema.dbml - -# 5. Setup environment cd output -echo "DATABASE_URL=sqlite+aiosqlite:///./database.db" > .env - -# 6. Install runtime dependencies (if not installed) -pip install dbml-to-sqlmodel[runtime] - -# 7. Run application -python main.py +echo "DATABASE_URL=sqlite+aiosqlite:///./db.db" > .env +python main.py # Run ``` -### Update Existing Application +### Updating an existing project ```bash -# 1. Modify schema.dbml (add new table/column) - -# 2. Preview changes +# After adding a column to schema.dbml dbml-to-sqlmodel preview schema.dbml -o my_app - -# 3. Review what will be updated -# Protected files (USER_MODIFIED) won't be touched - -# 4. Apply changes +# ✅ Shows only the changes dbml-to-sqlmodel generate schema.dbml -o my_app - -# 5. Restart application -cd my_app && python main.py ``` -### Multi-Environment Setup - -```bash -# Development -dbml-to-sqlmodel generate schema.dbml -o dev_app -cd dev_app -echo "DATABASE_URL=sqlite+aiosqlite:///./dev.db" > .env - -# Production -dbml-to-sqlmodel generate schema.dbml -o prod_app --admin-auth -cd prod_app -cat > .env < posts.id, note: "Reference to post"] - author_id integer [ref: > users.id, note: "Reference to user (null for anonymous)"] - parent_id integer [ref: > comments.id, note: "Reference to parent comment for nested comments"] + post_id integer [not null, note: "Reference to post"] + author_id integer [note: "Reference to user (null for anonymous)"] + parent_id integer [note: "Reference to parent comment for nested comments"] content text [not null, note: "Comment content"] - author_name2 text [note: "Name for anonymous comments"] + author_name text [note: "Name for anonymous comments"] author_email text [note: "Email for anonymous comments"] is_approved boolean [not null, default: False, note: "Is comment approved by moderator"] created_at timestamp [not null, note: "Comment creation timestamp"] diff --git a/pyproject.toml b/pyproject.toml index 5ef4ef0..ac04467 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,10 +1,46 @@ [project] name = "dbml-to-sqlmodel" -version = "0.1.0" -description = "Generate SQLModel models and FastAPI CRUD from DBML schema" +dynamic = ["version"] +description = "Generate FastAPI + SQLModel + FastCRUD + SQLAdmin applications from a DBML schema" readme = "README.md" license = "MIT" +license-files = ["LICENSE"] requires-python = ">=3.11" +authors = [ + { name = "azhig", email = "azhig.dev@gmail.com" }, +] +maintainers = [ + { name = "azhig", email = "azhig.dev@gmail.com" }, +] +keywords = [ + "dbml", + "sqlmodel", + "fastapi", + "fastcrud", + "sqladmin", + "code-generation", + "scaffold", + "crud", + "orm", + "database", +] +classifiers = [ + "Development Status :: 4 - Beta", + "Environment :: Console", + "Intended Audience :: Developers", + "License :: OSI Approved :: MIT License", + "Operating System :: OS Independent", + "Programming Language :: Python", + "Programming Language :: Python :: 3", + "Programming Language :: Python :: 3 :: Only", + "Programming Language :: Python :: 3.11", + "Programming Language :: Python :: 3.12", + "Programming Language :: Python :: 3.13", + "Topic :: Database", + "Topic :: Software Development :: Code Generators", + "Topic :: Internet :: WWW/HTTP :: HTTP Servers", + "Typing :: Typed", +] dependencies = [ "pydbml>=1.2.1", "typer>=0.17.0", @@ -14,11 +50,19 @@ dependencies = [ "unidiff>=0.7.5", ] +[project.urls] +Homepage = "https://github.com/azhig/dbml-to-sqlmodel" +Repository = "https://github.com/azhig/dbml-to-sqlmodel" +Documentation = "https://github.com/azhig/dbml-to-sqlmodel/blob/main/docs/CLI_GUIDE.md" +Changelog = "https://github.com/azhig/dbml-to-sqlmodel/blob/main/CHANGELOG.md" +Issues = "https://github.com/azhig/dbml-to-sqlmodel/issues" + [project.optional-dependencies] runtime = [ "aiosqlite>=0.22.1", "fastapi[all]>=0.128.0", "fastcrud>=0.20.1", + "greenlet>=3.0.0", "sqladmin>=0.22.0", "sqlmodel>=0.0.31", ] @@ -43,6 +87,25 @@ dbml2sm = "dbml_to_sqlmodel.cli:app" requires = ["hatchling"] build-backend = "hatchling.build" +[tool.hatch.version] +path = "src/dbml_to_sqlmodel/__init__.py" + +[tool.hatch.build.targets.wheel] +packages = ["src/dbml_to_sqlmodel"] + +[tool.hatch.build.targets.sdist] +include = [ + "src/dbml_to_sqlmodel", + "tests", + "examples", + "docs", + "README.md", + "CHANGELOG.md", + "CONTRIBUTING.md", + "LICENSE", + "Makefile", +] + [tool.pytest.ini_options] testpaths = ["tests"] python_files = ["test_*.py"] @@ -149,11 +212,3 @@ module = [ "unidiff.*", ] ignore_missing_imports = true - -[[tool.mypy.overrides]] -module = "dbml_to_sqlmodel.sqlmodel_to_dbml" -disable_error_code = ["assignment", "arg-type"] - -[[tool.mypy.overrides]] -module = "dbml_to_sqlmodel.generate_app" -ignore_errors = true diff --git a/src/__init__.py b/src/__init__.py deleted file mode 100644 index bc30b8e..0000000 --- a/src/__init__.py +++ /dev/null @@ -1,3 +0,0 @@ -"""DBML to Code Generator - FastAPI application generator from DBML schemas.""" - -__version__ = "0.1.0" diff --git a/src/dbml_to_sqlmodel/cli.py b/src/dbml_to_sqlmodel/cli.py index 9d3545e..58572ec 100644 --- a/src/dbml_to_sqlmodel/cli.py +++ b/src/dbml_to_sqlmodel/cli.py @@ -1,6 +1,5 @@ """Interactive CLI for DBML to SQLModel Generator.""" -import sys from pathlib import Path from typing import Annotated @@ -11,13 +10,11 @@ from rich.panel import Panel from rich.table import Table +from . import __version__ from .commands import code_to_dbml, generate, info, preview from .core import ConfigManager from .sqlmodel_to_dbml import apply_dbml_table_updates -# Add parent directory to path to import from src -sys.path.insert(0, str(Path(__file__).parent.parent)) - console = Console() config_manager = ConfigManager() @@ -379,8 +376,26 @@ def handle_code_to_dbml(): # pragma: no cover console.print("[green]DBML saved[/green]") +def _version_callback(value: bool) -> None: + if value: + console.print(f"dbml-to-sqlmodel {__version__}") + raise typer.Exit() + + @cli.callback(invoke_without_command=True) -def main(ctx: typer.Context): +def main( + ctx: typer.Context, + version: Annotated[ + bool | None, + typer.Option( + "--version", + "-V", + help="Show the version and exit.", + callback=_version_callback, + is_eager=True, + ), + ] = None, +): """DBML to SQLModel Generator - interactive mode or subcommands.""" if ctx.invoked_subcommand is None: # No subcommand provided, run interactive menu diff --git a/src/dbml_to_sqlmodel/constants.py b/src/dbml_to_sqlmodel/constants.py index 6edd841..13b8651 100644 --- a/src/dbml_to_sqlmodel/constants.py +++ b/src/dbml_to_sqlmodel/constants.py @@ -1,25 +1,11 @@ """Project-wide constants.""" -# Default paths -DEFAULT_SCHEMA_PATH = "examples/schema.dbml" -DEFAULT_OUTPUT_DIR = "output" +# Configuration file name (stored in the working directory) CONFIG_FILE = ".dbml_to_sqlmodel" -# File markers +# Marker used to protect manually edited generated files from being overwritten USER_FILE_MARKER = "# USER_MODIFIED" PROTECTED_FILE_WARNING = """# USER_MODIFIED # This file has been manually modified and is protected from regeneration. # Remove this marker if you want to allow regeneration. """ - -# Server defaults -DEFAULT_SERVER_PORT = 8001 -DEFAULT_SERVER_HOST = "0.0.0.0" - -# Admin panel -ADMIN_PATH = "/admin" -ADMIN_TITLE = "Admin Panel" - -# Database -DEFAULT_DB_NAME = "app.db" -DEFAULT_DB_URL_TEMPLATE = "sqlite:///{db_name}" diff --git a/src/dbml_to_sqlmodel/core/code_generator.py b/src/dbml_to_sqlmodel/core/code_generator.py index 4ae5bac..6bbe13c 100644 --- a/src/dbml_to_sqlmodel/core/code_generator.py +++ b/src/dbml_to_sqlmodel/core/code_generator.py @@ -3,7 +3,7 @@ from typing import Any from ..models.schema import TableInfo -from ..utils.type_mapping import get_python_type +from ..utils.type_mapping import get_datetime_imports, get_python_type def to_class_name(name: str) -> str: @@ -11,6 +11,21 @@ def to_class_name(name: str) -> str: return "".join(part.capitalize() for part in name.split("_") if part) +def _singularize(name: str) -> str: + """Best-effort singularization for relationship attribute names. + + Handles the common English plural endings without corrupting singular + nouns that happen to end in ``s`` (e.g. ``status`` stays ``status``). + """ + if name.endswith("ies") and len(name) > 3: + return name[:-3] + "y" + if name.endswith(("ses", "xes", "zes", "ches", "shes")): + return name[:-2] + if name.endswith("s") and not name.endswith(("ss", "us", "is")): + return name[:-1] + return name + + def _format_default_value(value: Any) -> str: """Format default value for Python code.""" if isinstance(value, bool): @@ -43,24 +58,32 @@ def generate_single_model( if column.note: field_descriptions[column.name] = column.note - # Collect relationships for this table + # Collect relationships for this table. + # related_tables holds the *snake_case* target table names (used for both the + # sibling module path and, via to_class_name, the referenced class name). table_relationships = [] - related_models = set() # Track which models need to be imported + related_tables: set[str] = set() enum_types = {to_class_name(col.type) for col in table.columns if col.type in enums} for column in table.columns: if column.references: target_table, target_col = column.references[0] if target_table != table.name or target_col != column.name: - rel_name = target_table.rstrip("s") + rel_name = _singularize(target_table) table_relationships.append((column.name, rel_name, target_table)) - related_models.add(target_table.capitalize()) + related_tables.add(target_table) # Check if uuid is needed (for string primary keys) needs_uuid = any( col.primary_key and get_python_type(col.type) == "str" for col in table.columns ) + # Standard-library datetime imports needed for temporal columns + used_python_types = { + get_python_type(col.type) for col in table.columns if col.type not in enums + } + datetime_imports = get_datetime_imports(used_python_types) + # Generate imports with TYPE_CHECKING block sqlmodel_imports = ["SQLModel", "Field"] if table_relationships: @@ -69,18 +92,19 @@ def generate_single_model( imports = f"""from sqlmodel import {", ".join(sqlmodel_imports)} from typing import Optional, TYPE_CHECKING """ + if datetime_imports: + imports += f"from datetime import {', '.join(datetime_imports)}\n" if enum_types: imports += f"from ..enums import {', '.join(sorted(enum_types))}\n" if needs_uuid: imports += "import uuid\n" imports += "\n" - if related_models: + if related_tables: imports += "if TYPE_CHECKING:\n" - for model in sorted(related_models): - # Import from sibling module - model_module = model.lower() - imports += f" from ..{model_module}.model import {model}\n" + for target_table in sorted(related_tables): + # Import the related model class from its sibling module + imports += f" from ..{target_table}.model import {to_class_name(target_table)}\n" imports += "\n" else: imports += """if TYPE_CHECKING: @@ -104,7 +128,8 @@ def generate_single_model( model_code = [] # Generate Base class with common fields - base_class_name = f"{table.name.capitalize()}Base" + table_class_name = to_class_name(table.name) + base_class_name = f"{table_class_name}Base" base_class_def = [f"class {base_class_name}(SQLModel):"] has_base_fields = False @@ -156,7 +181,6 @@ def generate_single_model( model_code.append("\n".join(base_class_def) + "\n") # Generate main table model (inherits from Base) - table_class_name = table.name.capitalize() class_def = [f"class {table_class_name}({base_class_name}, table=True):"] if table.note: note_escaped = table.note.replace('"""', '\\"\\"\\"') @@ -199,7 +223,7 @@ def generate_single_model( else: class_def.append("") for fk_col, rel_name, target_table in table_relationships: - target_class = target_table.capitalize() + target_class = to_class_name(target_table) is_nullable = any(c.name == fk_col and c.nullable for c in table.columns) rel_type = f'Optional["{target_class}"]' if is_nullable else f'"{target_class}"' class_def.append(f" {rel_name}: {rel_type} = Relationship()") diff --git a/src/dbml_to_sqlmodel/core/config.py b/src/dbml_to_sqlmodel/core/config.py index d77758f..a4fbea7 100644 --- a/src/dbml_to_sqlmodel/core/config.py +++ b/src/dbml_to_sqlmodel/core/config.py @@ -3,6 +3,8 @@ import json from pathlib import Path +from pydantic import ValidationError + from ..constants import CONFIG_FILE from ..models.config_models import AppConfig @@ -25,8 +27,8 @@ def load(self) -> AppConfig: try: data = json.loads(self.config_path.read_text(encoding="utf-8")) self._config = AppConfig(**data) - except Exception: - # If config is corrupted, create new one + except (json.JSONDecodeError, ValidationError, TypeError, OSError): + # If the config is corrupted or unreadable, recreate the default one self._config = AppConfig() self.save() else: diff --git a/src/dbml_to_sqlmodel/core/parser.py b/src/dbml_to_sqlmodel/core/parser.py index bf27215..492abe6 100644 --- a/src/dbml_to_sqlmodel/core/parser.py +++ b/src/dbml_to_sqlmodel/core/parser.py @@ -2,8 +2,11 @@ import re import warnings +from collections.abc import Iterator from typing import Any +from ..models.schema import ColumnInfo, RelationshipInfo, TableInfo + # Suppress pyparsing deprecation warnings from pydbml library # pydbml uses old pyparsing API (camelCase) which triggers deprecation warnings # in pyparsing 3.3+. These warnings occur during module import when pydbml @@ -29,10 +32,6 @@ def PyDBML(content: str) -> Any: return _PyDBML(content) -from ..integrations import setdefaultattr -from ..models.schema import ColumnInfo, RelationshipInfo, TableInfo - - def parse_dbml(dbml_content: str) -> list[TableInfo]: """Parse DBML content using PyDBML into structured table definitions.""" raw_types = extract_dbml_types(dbml_content) @@ -40,35 +39,28 @@ def parse_dbml(dbml_content: str) -> list[TableInfo]: parsed = PyDBML(dbml_content) tables = [] - # Process references first to handle foreign keys - for ref in parsed.refs: # type: ignore - if ref.type == ">": - for col in ref.col1: - col_refs = setdefaultattr(col, "references", []) - col_refs.append(ref) - elif ref.type == "<": - for col in ref.col2: - col_refs = setdefaultattr(col, "references", []) - col_refs.append(ref) - - for table in parsed.tables: # type: ignore + # Map each source column to the foreign-key refs that originate from it, + # keyed by object identity so the PyDBML objects are never mutated. + column_refs: dict[int, list] = {} + for ref in parsed.refs: + source_cols = ref.col1 if ref.type == ">" else ref.col2 if ref.type == "<" else [] + for col in source_cols: + column_refs.setdefault(id(col), []).append(ref) + + for table in parsed.tables: columns = [] for column in table.columns: col_type = column.type if hasattr(col_type, "name"): - col_type = col_type.name # type: ignore + col_type = col_type.name # Handle references references = [] - if hasattr(column, "references"): - for ref in column.references: # type: ignore - if ref.type == ">": - target_table = ref.table2.name - target_col = ref.col2[0].name - else: - target_table = ref.table1.name - target_col = ref.col1[0].name - references.append((target_table, target_col)) + for ref in column_refs.get(id(column), []): + if ref.type == ">": + references.append((ref.table2.name, ref.col2[0].name)) + else: + references.append((ref.table1.name, ref.col1[0].name)) # Extract note note_text = None @@ -131,9 +123,14 @@ def parse_dbml(dbml_content: str) -> list[TableInfo]: return tables -def extract_dbml_types(dbml_content: str) -> dict[str, dict[str, str]]: - """Extract raw column types from DBML text for stable round-tripping.""" - table_types: dict[str, dict[str, str]] = {} +def _iter_column_lines(dbml_content: str) -> Iterator[tuple[str, str]]: + """Yield ``(table_name, column_line)`` for each column definition in the text. + + Comments are stripped and table headers / notes / indexes / refs are skipped. + This lightweight scan recovers raw types and defaults that PyDBML normalizes + away, which is needed for stable round-tripping. Shared by + :func:`extract_dbml_types` and :func:`extract_dbml_defaults`. + """ current_table: str | None = None pending_table: str | None = None @@ -144,18 +141,15 @@ def extract_dbml_types(dbml_content: str) -> dict[str, dict[str, str]]: table_match = re.match(r"(?i)^table\s+([A-Za-z_][\w]*)", stripped) if table_match and current_table is None: - table_name = table_match.group(1) if "{" in stripped: - current_table = table_name - table_types.setdefault(current_table, {}) + current_table = table_match.group(1) else: - pending_table = table_name + pending_table = table_match.group(1) continue if pending_table and "{" in stripped: current_table = pending_table pending_table = None - table_types.setdefault(current_table, {}) continue if current_table: @@ -163,95 +157,53 @@ def extract_dbml_types(dbml_content: str) -> dict[str, dict[str, str]]: current_table = None continue - lowered = stripped.lower() - if ( - lowered.startswith("note:") - or lowered.startswith("indexes") - or lowered.startswith("ref") - ): + if stripped.lower().startswith(("note:", "indexes", "ref")): continue line_no_comment = stripped.split("//", 1)[0].strip() if not line_no_comment: # pragma: no cover continue - parts = line_no_comment.split("[", 1)[0].strip() - tokens = parts.split() - if len(tokens) < 2: - continue + yield current_table, line_no_comment - col_name = tokens[0] - col_type = " ".join(tokens[1:]) - table_types[current_table][col_name] = col_type +def extract_dbml_types(dbml_content: str) -> dict[str, dict[str, str]]: + """Extract raw column types from DBML text for stable round-tripping.""" + table_types: dict[str, dict[str, str]] = {} + for table_name, line in _iter_column_lines(dbml_content): + table_types.setdefault(table_name, {}) + tokens = line.split("[", 1)[0].split() + if len(tokens) < 2: + continue + table_types[table_name][tokens[0]] = " ".join(tokens[1:]) return table_types def extract_dbml_defaults(dbml_content: str) -> dict[str, dict[str, Any]]: """Extract raw defaults from DBML text.""" table_defaults: dict[str, dict[str, Any]] = {} - current_table: str | None = None - pending_table: str | None = None - - for line in dbml_content.splitlines(): - stripped = line.strip() - if not stripped or stripped.startswith("//"): + for table_name, line in _iter_column_lines(dbml_content): + table_defaults.setdefault(table_name, {}) + if "[" not in line or "]" not in line: continue - table_match = re.match(r"(?i)^table\s+([A-Za-z_][\w]*)", stripped) - if table_match and current_table is None: - table_name = table_match.group(1) - if "{" in stripped: - current_table = table_name - table_defaults.setdefault(current_table, {}) - else: - pending_table = table_name + before_attrs, attrs = line.split("[", 1) + attrs = attrs.split("]", 1)[0] + tokens = before_attrs.split() + if len(tokens) < 2: continue - if pending_table and "{" in stripped: - current_table = pending_table - pending_table = None - table_defaults.setdefault(current_table, {}) - continue - - if current_table: - if stripped.startswith("}"): - current_table = None - continue - - lowered = stripped.lower() - if ( - lowered.startswith("note:") - or lowered.startswith("indexes") - or lowered.startswith("ref") - ): + col_name = tokens[0] + for part in attrs.split(","): + if "default" not in part: continue - - line_no_comment = stripped.split("//", 1)[0].strip() - if not line_no_comment: # pragma: no cover + key, sep, value = part.partition(":") + if not sep: + key, sep, value = part.partition("=") + if key.strip() != "default": continue - - if "[" not in line_no_comment or "]" not in line_no_comment: - continue - - before_attrs, attrs = line_no_comment.split("[", 1) - attrs = attrs.split("]", 1)[0] - tokens = before_attrs.split() - if len(tokens) < 2: - continue - - col_name = tokens[0] - for part in attrs.split(","): - if "default" not in part: - continue - key, _, value = part.partition(":") - if not _: - key, _, value = part.partition("=") - if key.strip() != "default": - continue - default_value = _parse_default_value(value) - table_defaults[current_table][col_name] = default_value - break + table_defaults[table_name][col_name] = _parse_default_value(value) + break return table_defaults diff --git a/src/dbml_to_sqlmodel/generate_app.py b/src/dbml_to_sqlmodel/generate_app.py deleted file mode 100644 index b066616..0000000 --- a/src/dbml_to_sqlmodel/generate_app.py +++ /dev/null @@ -1,391 +0,0 @@ -import argparse -import os -import re -import sys - -from .core import parse_dbml, parse_dbml_enums -from .generator import generate_enums_file - - -def generate_crud_router(table): - """Generate CRUD router for a single table""" - model_name = table.name.capitalize() - router_name = f"{table.name}_router" - - code = f"""from fastcrud import crud_router -from .model import {model_name}, {model_name}Create, {model_name}Update - - -def create_{table.name}_router(get_session_func): - \"\"\"Create CRUD router with session dependency\"\"\" - return crud_router( - model={model_name}, - create_schema={model_name}Create, - update_schema={model_name}Update, - path='/{table.name}', - tags=['{table.name}'], - session=get_session_func - ) - - -# This will be initialized in main.py -{router_name} = None -""" - return code - - -def generate_admin_views(tables, admin_auth_enabled=False): - imports = "\n".join( - [f"from models.{table.name} import {table.name.capitalize()}" for table in tables] - ) - - if admin_auth_enabled: - admin_imports = """import os - -from sqladmin import Admin, ModelView -from sqladmin.authentication import AuthenticationBackend -from starlette.middleware.sessions import SessionMiddleware -""" - auth_block = """class AdminAuth(AuthenticationBackend): - async def login(self, request): - form = await request.form() - if ( - form.get("username") == os.getenv("ADMIN_USER") - and form.get("password") == os.getenv("ADMIN_PASS") - ): - request.session.update({"admin": True}) - return True - return False - - async def logout(self, request): - request.session.clear() - return True - - async def authenticate(self, request): - return bool(request.session.get("admin")) - -""" - init_admin_header = """def init_admin(app, engine): - app.add_middleware(SessionMiddleware, secret_key=os.getenv("ADMIN_SECRET", "change-me")) - auth = AdminAuth(secret_key=os.getenv("ADMIN_SECRET", "change-me")) - admin = Admin(app, engine, authentication_backend=auth) - -""" - else: - admin_imports = "from sqladmin import Admin, ModelView\n" - auth_block = "" - init_admin_header = """def init_admin(app, engine): - admin = Admin(app, engine) - -""" - - code = f"""{admin_imports} - -{imports} - -# Default icons for different table types -DEFAULT_ICONS = {{ - 'user': 'fa-solid fa-user', - 'product': 'fa-solid fa-box', - 'order': 'fa-solid fa-shopping-cart', - 'team': 'fa-solid fa-users', - 'skill': 'fa-solid fa-star', - 'scenario': 'fa-solid fa-diagram-project', - 'function': 'fa-solid fa-code', - 'status': 'fa-solid fa-circle-check', - 'stage': 'fa-solid fa-layer-group', - 'action': 'fa-solid fa-bolt', - 'default': 'fa-solid fa-table' -}} - -def get_icon_for_table(table_name): - \"\"\"Select icon based on table name\"\"\" - name_lower = table_name.lower() - for keyword, icon in DEFAULT_ICONS.items(): - if keyword in name_lower: - return icon - return DEFAULT_ICONS['default'] - -{auth_block}{init_admin_header}""" - for table in tables: - model_name = table.name.capitalize() - - code += f" class {model_name}Admin(ModelView, model={model_name}):\n" - code += f" name = '{model_name}'\n" - code += f" name_plural = '{model_name}'\n" - code += f" icon = get_icon_for_table('{table.name}')\n" - code += f" column_list = [c.name for c in {model_name}.__table__.columns]\n" - - # Add column_labels from notes - labels = {c.name: c.note for c in table.columns if c.note} - if labels: - code += " column_labels = {\n" - for col_name, note in labels.items(): - note_escaped = note.replace("'", "\\'") - code += f" '{col_name}': '{note_escaped}',\n" - code += " }\n" - - # Enable sorting for all columns - sortable_columns = [c.name for c in table.columns] - code += f" column_sortable_list = {sortable_columns}\n" - - # Add search for text fields - text_types = ["varchar", "text", "string", "str", "char"] - searchable_columns = [ - c.name for c in table.columns if any(t in c.type.lower() for t in text_types) - ] - if searchable_columns: - code += f" column_searchable_list = {searchable_columns}\n" - - code += f"\n admin.add_view({model_name}Admin)\n\n" - - code += " return admin\n" - return code - - -def _extract_enums_from_text(dbml_content: str) -> dict[str, list[str]]: - enums: dict[str, list[str]] = {} - current: str | None = None - - for line in dbml_content.splitlines(): - stripped = line.strip() - if not stripped or stripped.startswith("//"): - continue - - enum_match = re.match(r"(?i)^enum\s+([A-Za-z_][\w]*)", stripped) - if enum_match and current is None: - enum_name = enum_match.group(1) - enums.setdefault(enum_name, []) - current = enum_name - continue - - if current: - if "}" in stripped: - current = None - continue - value = stripped.split("//", 1)[0].strip().strip(",") - if value: - enums[current].append(value) - - return enums - - -def generate_main_app(tables): - router_imports = "\n".join( - [f"from models.{table.name} import create_{table.name}_router" for table in tables] - ) - router_creates = "\n".join( - [f"{table.name}_router = create_{table.name}_router(get_session)" for table in tables] - ) - router_includes = "\n".join([f"app.include_router({table.name}_router)" for table in tables]) - - return f"""from fastapi import FastAPI -from sqlalchemy.ext.asyncio import create_async_engine -from sqlmodel.ext.asyncio.session import AsyncSession -from sqlmodel import SQLModel -from typing import AsyncGenerator -from dotenv import load_dotenv - - -load_dotenv() - - -DATABASE_URL = "sqlite+aiosqlite:///./database.db" - -# Database engine with basic configuration suitable for development -# For production deployments, consider tuning these parameters: -# pool_size - number of connections to keep open (default: 5) -# max_overflow - max connections beyond pool_size (default: 10) -# pool_pre_ping - verify connections before using (recommended for production) -# pool_recycle - recycle connections after N seconds (e.g., 3600) -# echo - set to False in production to reduce logging -# Example for production: -# engine = create_async_engine( -# DATABASE_URL, -# echo=False, -# pool_size=20, -# max_overflow=10, -# pool_pre_ping=True, -# pool_recycle=3600 -# ) -engine = create_async_engine(DATABASE_URL, echo=False) - - -# Define dependency for session -async def get_session() -> AsyncGenerator[AsyncSession, None]: - async with AsyncSession(engine) as session: - yield session - - -app = FastAPI() - - -# Create database tables -async def init_db(): - async with engine.begin() as conn: - await conn.run_sync(SQLModel.metadata.create_all) - - -@app.on_event("startup") -async def startup(): - await init_db() - - -# Import router factories AFTER get_session is defined -{router_imports} -from admin import init_admin - -# Create CRUD routers with session dependency -{router_creates} - -# Include all CRUD routers -{router_includes} - -# Mount admin panel -init_admin(app, engine) - - -@app.get("/") -def read_root(): - return {{"message": "FastAPI app generated from DBML"}} - - -if __name__ == "__main__": - import uvicorn - uvicorn.run(app, host="0.0.0.0", port=8001) -""" - - -def generate_requirements(): - return """fastapi>=0.104.0 -sqlmodel[async]>=0.0.14 -sqladmin>=1.4.0 -fastcrud>=0.6.0 -uvicorn[standard]>=0.23.0 -aiosqlite>=0.19.0 -python-dotenv>=1.0.0 -""" - - -def generate_model_init(table): - """Generate __init__.py for model directory""" - model_name = table.name.capitalize() - return f"""from .model import {model_name}, {model_name}Create, {model_name}Update -from .crud import create_{table.name}_router - -__all__ = [ - "{model_name}", - "{model_name}Create", - "{model_name}Update", - "create_{table.name}_router", -] -""" - - -def generate_models_root_init(tables): - """Generate root __init__.py for models directory""" - imports = [] - all_exports = [] - - for table in tables: - model_name = table.name.capitalize() - imports.append( - f"from .{table.name} import {model_name}, {model_name}Create, {model_name}Update, create_{table.name}_router" - ) - all_exports.extend( - [ - f' "{model_name}"', - f' "{model_name}Create"', - f' "{model_name}Update"', - f' "create_{table.name}_router"', - ] - ) - - return "\n".join(imports) + "\n\n__all__ = [\n" + ",\n".join(all_exports) + "\n]\n" - - -def main(): - parser = argparse.ArgumentParser(description="Generate FastAPI app from DBML") - parser.add_argument("dbml_file", type=str, help="Path to DBML file") - parser.add_argument( - "-o", - "--output", - type=str, - default="output", - help="Output directory for generated files (default: output)", - ) - parser.add_argument( - "--admin-auth", - action="store_true", - help="Require login for SQLAdmin panel", - ) - args = parser.parse_args() - - with open(args.dbml_file) as f: - dbml_content = f.read() - - tables = parse_dbml(dbml_content) - enums = parse_dbml_enums(dbml_content) - if not enums: - enums = _extract_enums_from_text(dbml_content) - if not tables: - print("Error: No tables found in DBML file") - sys.exit(1) - - # Create output directory - output_dir = args.output - os.makedirs(output_dir, exist_ok=True) - - # Create models directory - models_dir = os.path.join(output_dir, "models") - os.makedirs(models_dir, exist_ok=True) - - # Generate each model in its own subdirectory - from dbml_to_sqlmodel import generate_single_model - - for table in tables: - # Create model subdirectory - model_dir = os.path.join(models_dir, table.name) - os.makedirs(model_dir, exist_ok=True) - - # Create model.py - with open(os.path.join(model_dir, "model.py"), "w") as f: - f.write(generate_single_model(table, tables, enums=enums)) - - # Create crud.py - with open(os.path.join(model_dir, "crud.py"), "w") as f: - f.write(generate_crud_router(table)) - - # Create __init__.py - with open(os.path.join(model_dir, "__init__.py"), "w") as f: - f.write(generate_model_init(table)) - - # Create root models __init__.py - with open(os.path.join(models_dir, "__init__.py"), "w") as f: - f.write(generate_models_root_init(tables)) - - if enums: - with open(os.path.join(models_dir, "enums.py"), "w") as f: - f.write(generate_enums_file(enums)) - - # Create admin.py - with open(os.path.join(output_dir, "admin.py"), "w") as f: - f.write(generate_admin_views(tables, admin_auth_enabled=args.admin_auth)) - - # Create main.py - with open(os.path.join(output_dir, "main.py"), "w") as f: - f.write(generate_main_app(tables)) - - print(f"Successfully generated FastAPI app with {len(tables)} models") - print(f"Output directory: {output_dir}") - print(f"Structure: models/{'{table_name}'}/(model.py, crud.py, __init__.py)") - - # Create requirements.txt - with open(os.path.join(output_dir, "requirements.txt"), "w") as f: - f.write(generate_requirements()) - - print("Files created: requirements.txt") - - -if __name__ == "__main__": # pragma: no cover - main() diff --git a/src/dbml_to_sqlmodel/generator.py b/src/dbml_to_sqlmodel/generator.py index d5bcaad..27e0a4a 100644 --- a/src/dbml_to_sqlmodel/generator.py +++ b/src/dbml_to_sqlmodel/generator.py @@ -1,14 +1,12 @@ """Generator functions for creating FastAPI application files.""" -import re - from .core import generate_single_model, parse_dbml, parse_dbml_enums, to_class_name from .models import TableInfo def generate_crud_router(table: TableInfo) -> str: """Generate CRUD router for a single table.""" - model_name = table.name.capitalize() + model_name = to_class_name(table.name) router_name = f"{table.name}_router" return f"""from fastcrud import crud_router @@ -35,11 +33,12 @@ def create_{table.name}_router(get_session_func): def generate_admin_views(tables: list[TableInfo], admin_auth_enabled: bool = False) -> str: """Generate SQLAdmin views for all tables.""" imports = "\n".join( - [f"from models.{table.name} import {table.name.capitalize()}" for table in tables] + [f"from models.{table.name} import {to_class_name(table.name)}" for table in tables] ) if admin_auth_enabled: admin_imports = """import os +import secrets from sqladmin import Admin, ModelView from sqladmin.authentication import AuthenticationBackend @@ -48,9 +47,13 @@ def generate_admin_views(tables: list[TableInfo], admin_auth_enabled: bool = Fal auth_block = """class AdminAuth(AuthenticationBackend): async def login(self, request): form = await request.form() - if ( - form.get("username") == os.getenv("ADMIN_USER") - and form.get("password") == os.getenv("ADMIN_PASS") + username = str(form.get("username") or "") + password = str(form.get("password") or "") + expected_user = os.getenv("ADMIN_USER") or "" + expected_pass = os.getenv("ADMIN_PASS") or "" + # Constant-time comparison to avoid leaking credentials via timing + if secrets.compare_digest(username, expected_user) and secrets.compare_digest( + password, expected_pass ): request.session.update({"admin": True}) return True @@ -107,7 +110,7 @@ def get_icon_for_table(table_name): {auth_block}{init_admin_header}""" for table in tables: - model_name = table.name.capitalize() + model_name = to_class_name(table.name) code += f" class {model_name}Admin(ModelView, model={model_name}):\n" code += f" name = '{model_name}'\n" @@ -152,20 +155,24 @@ def generate_main_app(tables: list[TableInfo]) -> str: ) router_includes = "\n".join([f"app.include_router({table.name}_router)" for table in tables]) - return f"""from fastapi import FastAPI + return f"""import os +from contextlib import asynccontextmanager +from typing import AsyncGenerator + +from fastapi import FastAPI from sqlalchemy.ext.asyncio import create_async_engine -from sqlmodel.ext.asyncio.session import AsyncSession from sqlmodel import SQLModel -from typing import AsyncGenerator +from sqlmodel.ext.asyncio.session import AsyncSession from dotenv import load_dotenv load_dotenv() -DATABASE_URL = "sqlite+aiosqlite:///./database.db" +# Read the database URL from the environment (.env), with a sensible default. +DATABASE_URL = os.getenv("DATABASE_URL", "sqlite+aiosqlite:///./database.db") -# Database engine with basic configuration suitable for development +# Database engine with basic configuration suitable for development. # For production deployments, consider tuning these parameters: # pool_size - number of connections to keep open (default: 5) # max_overflow - max connections beyond pool_size (default: 10) @@ -190,24 +197,24 @@ async def get_session() -> AsyncGenerator[AsyncSession, None]: yield session -app = FastAPI() - - -# Create database tables -async def init_db(): +# Create database tables on startup using the modern lifespan API +async def init_db() -> None: async with engine.begin() as conn: await conn.run_sync(SQLModel.metadata.create_all) -@app.on_event("startup") -async def startup(): +@asynccontextmanager +async def lifespan(app: FastAPI): await init_db() + yield # Import router factories AFTER get_session is defined {router_imports} from admin import init_admin +app = FastAPI(lifespan=lifespan) + # Create CRUD routers with session dependency {router_creates} @@ -231,19 +238,20 @@ def read_root(): def generate_requirements() -> str: """Generate requirements.txt content.""" - return """fastapi>=0.104.0 -sqlmodel[async]>=0.0.14 -sqladmin>=1.4.0 -fastcrud>=0.6.0 -uvicorn[standard]>=0.23.0 -aiosqlite>=0.19.0 + return """fastapi[all]>=0.128.0 +sqlmodel>=0.0.31 +sqladmin>=0.22.0 +fastcrud>=0.20.1 +uvicorn[standard]>=0.30.0 +aiosqlite>=0.20.0 python-dotenv>=1.0.0 +greenlet>=3.0.0 """ def generate_model_init(table: TableInfo) -> str: """Generate __init__.py for model directory.""" - model_name = table.name.capitalize() + model_name = to_class_name(table.name) return f"""from .model import {model_name}, {model_name}Create, {model_name}Update from .crud import create_{table.name}_router @@ -262,7 +270,7 @@ def generate_models_root_init(tables: list[TableInfo]) -> str: all_exports = [] for table in tables: - model_name = table.name.capitalize() + model_name = to_class_name(table.name) imports.append( f"from .{table.name} import {model_name}, {model_name}Create, {model_name}Update, create_{table.name}_router" ) @@ -295,33 +303,6 @@ def generate_enums_file(enums: dict[str, list[str]]) -> str: return "\n".join(lines).rstrip() + "\n" -def _extract_enums_from_text(dbml_content: str) -> dict[str, list[str]]: - enums: dict[str, list[str]] = {} - current: str | None = None - - for line in dbml_content.splitlines(): - stripped = line.strip() - if not stripped or stripped.startswith("//"): - continue - - enum_match = re.match(r"(?i)^enum\s+([A-Za-z_][\w]*)", stripped) - if enum_match and current is None: - enum_name = enum_match.group(1) - enums.setdefault(enum_name, []) - current = enum_name - continue - - if current: - if "}" in stripped: - current = None - continue - value = stripped.split("//", 1)[0].strip().strip(",") - if value: - enums[current].append(value) - - return enums - - def generate_all_files(dbml_content: str, admin_auth_enabled: bool = False) -> dict[str, str]: """ Generate all application files from DBML content. @@ -330,9 +311,9 @@ def generate_all_files(dbml_content: str, admin_auth_enabled: bool = False) -> d Dict mapping relative file paths to their content """ tables = parse_dbml(dbml_content) + # parse_dbml_enums already parses enums from the raw text first, then falls + # back to PyDBML, so no extra text-extraction pass is needed here. enums = parse_dbml_enums(dbml_content) - if not enums: - enums = _extract_enums_from_text(dbml_content) if not tables: raise ValueError("No tables found in DBML file") diff --git a/src/dbml_to_sqlmodel/integrations/__init__.py b/src/dbml_to_sqlmodel/integrations/__init__.py deleted file mode 100644 index ab4b030..0000000 --- a/src/dbml_to_sqlmodel/integrations/__init__.py +++ /dev/null @@ -1,5 +0,0 @@ -"""External integrations package.""" - -from .pydbml_adapter import setdefaultattr - -__all__ = ["setdefaultattr"] diff --git a/src/dbml_to_sqlmodel/integrations/pydbml_adapter.py b/src/dbml_to_sqlmodel/integrations/pydbml_adapter.py deleted file mode 100644 index 24c886e..0000000 --- a/src/dbml_to_sqlmodel/integrations/pydbml_adapter.py +++ /dev/null @@ -1,22 +0,0 @@ -"""PyDBML library adapter.""" - -from typing import Any - - -def setdefaultattr(obj: Any, name: str, value: Any) -> Any: - """Get attribute if exists, else set to default and get. - - This is a helper function for working with PyDBML objects - which may not have all attributes set. - - Args: - obj: Object to check/modify - name: Attribute name - value: Default value to set if attribute doesn't exist - - Returns: - Attribute value (existing or newly set default) - """ - if not hasattr(obj, name): - setattr(obj, name, value) - return getattr(obj, name) diff --git a/src/dbml_to_sqlmodel/sqlmodel_to_dbml.py b/src/dbml_to_sqlmodel/sqlmodel_to_dbml.py index 3192da5..fb871bd 100644 --- a/src/dbml_to_sqlmodel/sqlmodel_to_dbml.py +++ b/src/dbml_to_sqlmodel/sqlmodel_to_dbml.py @@ -17,17 +17,28 @@ "bool": "bool", "float": "float", "dict": "json", + "datetime": "timestamp", + "date": "date", + "time": "time", } _DBML_TYPE_EQUIVALENCE = { "integer": {"int", "integer", "serial", "serial4", "serial8", "bigserial", "bigint"}, - "text": {"text", "varchar", "string", "date", "datetime", "timestamp"}, + "text": {"text", "varchar", "string"}, "bool": {"bool", "boolean"}, "float": {"float", "double", "decimal"}, "json": {"json"}, + "timestamp": {"timestamp", "timestamptz", "datetime"}, + "date": {"date"}, + "time": {"time"}, } +def _as_optional_str(value: object) -> str | None: + """Narrow a parsed Field kwarg value to ``str | None``.""" + return value if isinstance(value, str) else None + + def _canonical_dbml_type(dbml_type: str) -> str: lower = dbml_type.lower() for canonical, values in _DBML_TYPE_EQUIVALENCE.items(): @@ -67,7 +78,7 @@ def _annotation_to_type(annotation: ast.expr) -> tuple[str, bool]: def _parse_field_kwargs( - call: ast.Call, dict_constants: dict[str, dict] = None + call: ast.Call, dict_constants: dict[str, dict] | None = None ) -> dict[str, object]: """Parse Field() keyword arguments. @@ -129,7 +140,7 @@ def _extract_dict_constants(tree: ast.Module) -> dict[str, dict]: def _extract_columns( - class_def: ast.ClassDef, dict_constants: dict[str, dict] = None + class_def: ast.ClassDef, dict_constants: dict[str, dict] | None = None ) -> list[_ParsedColumn]: """Extract columns from a ClassDef node. @@ -164,10 +175,10 @@ def _extract_columns( nullable=is_optional, primary_key=bool(field_kwargs.get("primary_key", False)), unique=bool(field_kwargs.get("unique", False)), - foreign_key=field_kwargs.get("foreign_key"), - description=field_kwargs.get("description"), + foreign_key=_as_optional_str(field_kwargs.get("foreign_key")), + description=_as_optional_str(field_kwargs.get("description")), default=field_kwargs.get("default"), - default_factory=field_kwargs.get("default_factory"), + default_factory=_as_optional_str(field_kwargs.get("default_factory")), ) ) return columns diff --git a/src/dbml_to_sqlmodel/templates/__init__.py b/src/dbml_to_sqlmodel/templates/__init__.py deleted file mode 100644 index e69de29..0000000 diff --git a/src/dbml_to_sqlmodel/utils/type_mapping.py b/src/dbml_to_sqlmodel/utils/type_mapping.py index b6954a9..5565b71 100644 --- a/src/dbml_to_sqlmodel/utils/type_mapping.py +++ b/src/dbml_to_sqlmodel/utils/type_mapping.py @@ -14,15 +14,20 @@ "string": "str", "bool": "bool", "boolean": "bool", - "date": "str", - "datetime": "str", - "timestamp": "str", + "date": "date", + "datetime": "datetime", + "timestamp": "datetime", + "timestamptz": "datetime", + "time": "time", "float": "float", "double": "float", "decimal": "float", "json": "dict", } +# Python types that require an import from the standard library ``datetime`` module. +DATETIME_TYPES: frozenset[str] = frozenset({"datetime", "date", "time"}) + def get_python_type(dbml_type: str) -> str: """Get Python type for a DBML type. @@ -36,6 +41,19 @@ def get_python_type(dbml_type: str) -> str: return TYPE_MAPPING.get(dbml_type.lower(), "str") +def get_datetime_imports(python_types: set[str]) -> list[str]: + """Return the sorted ``datetime`` names that must be imported for these types. + + Args: + python_types: Set of Python type names used in a generated module + + Returns: + Sorted list of names to import from the ``datetime`` module + (e.g. ``["date", "datetime"]``) + """ + return sorted(DATETIME_TYPES & python_types) + + def is_auto_increment_type(dbml_type: str) -> bool: """Check if DBML type is auto-incrementing. diff --git a/tests/test_cli.py b/tests/test_cli.py new file mode 100644 index 0000000..a7c4552 --- /dev/null +++ b/tests/test_cli.py @@ -0,0 +1,86 @@ +"""Tests for the typer CLI layer (cli.py).""" + +import sys +from pathlib import Path + +import pytest + +# Add src to path for imports +sys.path.insert(0, str(Path(__file__).parent.parent / "src")) + +from dbml_to_sqlmodel import cli as cli_module +from dbml_to_sqlmodel.commands import code_to_dbml, generate, info, preview + + +def test_cli_app_error_paths(monkeypatch): + # Make argv deterministic so cli() does not pick up pytest's own flags. + monkeypatch.setattr(sys, "argv", ["dbml-to-sqlmodel"]) + + # KeyboardInterrupt is handled inside the callback; the standalone app then + # exits cleanly with code 0. This also exercises the app() entry wrapper. + def raise_keyboard(): + raise KeyboardInterrupt + + monkeypatch.setattr(cli_module, "interactive_menu", raise_keyboard) + with pytest.raises(SystemExit) as exc_info: + cli_module.app() + assert exc_info.value.code in (0, None) + + # A generic exception propagates out of the callback (standalone_mode=False + # makes click re-raise instead of converting it into a sys.exit()). + def raise_error(): + raise ValueError("boom") + + monkeypatch.setattr(cli_module, "interactive_menu", raise_error) + with pytest.raises(ValueError): + cli_module.cli([], standalone_mode=False) + + +def test_cli_version_flag(): + """`--version` prints the version and exits with code 0.""" + with pytest.raises(SystemExit) as exc_info: + cli_module.cli(["--version"]) + assert exc_info.value.code == 0 + + +def test_cli_subcommand_wrappers(tmp_path, monkeypatch): + """The typer command wrappers forward their arguments to the command layer.""" + schema = tmp_path / "schema.dbml" + schema.write_text("Table users {\n id integer [pk]\n}\n", encoding="utf-8") + out = tmp_path / "out" + + calls: dict[str, dict] = {} + monkeypatch.setattr(generate, "generate_command", lambda **kw: calls.update(generate=kw)) + monkeypatch.setattr(preview, "preview_command", lambda **kw: calls.update(preview=kw)) + monkeypatch.setattr(info, "info_command", lambda **kw: calls.update(info=kw)) + + cli_module.cli(["generate", str(schema), "-o", str(out)], standalone_mode=False) + assert calls["generate"]["schema_file"] == schema + assert calls["generate"]["output"] == out + + cli_module.cli(["preview", str(schema), "-o", str(out)], standalone_mode=False) + assert calls["preview"]["schema_file"] == schema + + cli_module.cli(["info", str(schema), "-o", str(out)], standalone_mode=False) + assert calls["info"]["schema_file"] == schema + + +def test_cli_code_to_dbml_wrapper(tmp_path, monkeypatch): + """code-to-dbml writes only when changes are detected.""" + source_dir = tmp_path / "out" + source_dir.mkdir() + target = tmp_path / "schema.dbml" + + # No changes -> the file is not written. + monkeypatch.setattr( + code_to_dbml, "code_to_dbml_command", lambda **kw: ("dbml", "existing", False) + ) + cli_module.cli(["code-to-dbml", str(source_dir), "-o", str(target)], standalone_mode=False) + assert not target.exists() + + # Changes detected -> the DBML file is written. + generated = "Table users {\n id integer [pk]\n}\n" + monkeypatch.setattr(code_to_dbml, "code_to_dbml_command", lambda **kw: (generated, "", True)) + cli_module.cli(["code-to-dbml", str(source_dir), "-o", str(target)], standalone_mode=False) + assert target.exists() + assert "users" in target.read_text(encoding="utf-8") diff --git a/tests/test_code_generator.py b/tests/test_code_generator.py index d928ad1..57674e5 100644 --- a/tests/test_code_generator.py +++ b/tests/test_code_generator.py @@ -220,3 +220,87 @@ def test_generate_single_model_with_default_value(self): assert "active" in code assert "bool" in code + + +def test_generate_single_model_edge_cases(): + enum_table = TableInfo( + name="widgets", + columns=[ + ColumnInfo( + name="id", + type="bool", + primary_key=True, + unique=False, + nullable=False, + note="id note", + ), + ColumnInfo( + name="status", + type="status", + primary_key=True, + unique=False, + nullable=False, + ), + ], + note='note """ triple', + ) + rel_table = TableInfo( + name="posts", + columns=[ + ColumnInfo( + name="user_id", + type="integer", + primary_key=False, + unique=False, + nullable=False, + references=[("users", "id")], + ), + ], + ) + nopk_table = TableInfo( + name="logs", + columns=[ + ColumnInfo( + name="message", + type="text", + primary_key=False, + unique=False, + nullable=True, + ), + ], + ) + all_tables = [enum_table, rel_table, nopk_table] + enums = {"status": ["active"]} + + enum_code = generate_single_model(enum_table, all_tables, enums=enums) + assert '"""note \\"\\"\\" triple"""' in enum_code + assert "DESCRIPTIONS" in enum_code + assert 'description=DESCRIPTIONS["id"]' in enum_code + assert "status: Status" in enum_code + + rel_code = generate_single_model(rel_table, all_tables, enums=enums) + assert "Relationship()" in rel_code + + nopk_code = generate_single_model(nopk_table, all_tables, enums=enums) + assert "class Logs(LogsBase, table=True):\n pass" in nopk_code + + +def test_singularize_branches(): + """Every branch of the relationship-name singularizer is exercised.""" + from dbml_to_sqlmodel.core.code_generator import _singularize + + # "ies" -> "y" + assert _singularize("categories") == "category" + # short "ies" word is left untouched (len <= 3 guard) + assert _singularize("ies") == "ie" + # "ses"/"xes"/"ches"/"shes"/"zes" -> drop the trailing "es" + assert _singularize("statuses") == "status" + assert _singularize("boxes") == "box" + assert _singularize("batches") == "batch" + # plain trailing "s" -> drop it + assert _singularize("posts") == "post" + # singular nouns ending in ss/us/is are preserved (fallback branch) + assert _singularize("status") == "status" + assert _singularize("address") == "address" + assert _singularize("analysis") == "analysis" + assert _singularize("user") == "user" diff --git a/tests/test_commands.py b/tests/test_commands.py new file mode 100644 index 0000000..7dd9c9c --- /dev/null +++ b/tests/test_commands.py @@ -0,0 +1,158 @@ +"""Tests for the command layer (commands/*.py).""" + +import sys +from pathlib import Path + +import pytest +import typer + +# Add src to path for imports +sys.path.insert(0, str(Path(__file__).parent.parent / "src")) + +from dbml_to_sqlmodel.commands import code_to_dbml, generate, info, preview +from dbml_to_sqlmodel.constants import USER_FILE_MARKER +from dbml_to_sqlmodel.generator import generate_all_files + + +def _write_model(tmp_path: Path, rel_path: str, content: str) -> Path: + file_path = tmp_path / rel_path + file_path.parent.mkdir(parents=True, exist_ok=True) + file_path.write_text(content, encoding="utf-8") + return file_path + + +def _dbml_sample_tables() -> str: + return """ +Table users { + id integer [pk] + name text +} +Table teams { + id integer [pk] + name text +} +Table projects { + id integer [pk] + name text +} +Table notes { + id integer [pk] + body text +} +""" + + +def test_commands_preview_info_generate(tmp_path): + schema_file = tmp_path / "schema.dbml" + schema_file.write_text(_dbml_sample_tables(), encoding="utf-8") + + output = tmp_path / "out" + output.mkdir() + + generated_files = generate_all_files(schema_file.read_text(encoding="utf-8")) + + for rel_path, content in generated_files.items(): + if not rel_path.endswith("/model.py"): + continue + file_path = output / rel_path + file_path.parent.mkdir(parents=True, exist_ok=True) + if "users" in rel_path: + file_path.write_text(content, encoding="utf-8") + elif "teams" in rel_path: + file_path.write_text(content + "\nEXTRA = 1\n", encoding="utf-8") + elif "projects" in rel_path: + file_path.write_text(USER_FILE_MARKER + "\n" + content, encoding="utf-8") + else: + pass + + summary = preview.preview_command( + schema_file=schema_file, output=output, show_all=True, show_new=True + ) + assert summary["new"] >= 1 + assert summary["modified"] >= 1 + assert summary["protected"] >= 1 + + preview.preview_command(schema_file=schema_file, output=output, show_all=False, show_new=False) + + info.info_command(schema_file=schema_file, output=output) + + generate.generate_command( + schema_file=schema_file, output=output, force=False, admin_auth_enabled=False + ) + + generate.generate_command( + schema_file=schema_file, output=output, force=True, admin_auth_enabled=True + ) + + +def test_commands_generate_all_files_errors(tmp_path, monkeypatch): + schema_file = tmp_path / "schema.dbml" + schema_file.write_text("Table users {\n id integer [pk]\n}\n", encoding="utf-8") + + def raise_generate(*_args, **_kwargs): + raise RuntimeError("boom") + + monkeypatch.setattr(preview, "generate_all_files", raise_generate) + with pytest.raises(typer.Exit): + preview.preview_command(schema_file=schema_file, output=tmp_path) + + monkeypatch.setattr(info, "generate_all_files", raise_generate) + with pytest.raises(typer.Exit): + info.info_command(schema_file=schema_file, output=tmp_path) + + monkeypatch.setattr(generate, "generate_all_files", raise_generate) + with pytest.raises(typer.Exit): + generate.generate_command(schema_file=schema_file, output=tmp_path) + + +def test_commands_error_paths(tmp_path, monkeypatch): + schema_file = tmp_path / "schema.dbml" + schema_file.write_text("Table users { id integer [pk] }", encoding="utf-8") + + def bad_read(*_args, **_kwargs): + raise OSError("boom") + + monkeypatch.setattr(Path, "read_text", bad_read) + + with pytest.raises(typer.Exit): + preview.preview_command(schema_file=schema_file) + + with pytest.raises(typer.Exit): + info.info_command(schema_file=schema_file) + + with pytest.raises(typer.Exit): + generate.generate_command(schema_file=schema_file) + + +def test_code_to_dbml_command_branches(tmp_path): + schema_file = tmp_path / "schema.dbml" + + with pytest.raises(Exception): + code_to_dbml.code_to_dbml_command(schema_file=schema_file, output=tmp_path) + + models_dir = tmp_path / "out" / "models" + models_dir.mkdir(parents=True) + _write_model(tmp_path, "out/models/enums.py", "from enum import Enum\n") + model_py = """ +from sqlmodel import Field, SQLModel + +class User(SQLModel, table=True): + id: int = Field(primary_key=True) +""" + _write_model(tmp_path, "out/models/user/model.py", model_py) + + generated, existing, changed = code_to_dbml.code_to_dbml_command( + schema_file=schema_file, + output=tmp_path / "out", + ) + assert existing == "" + assert changed is True + assert "Table user" in generated + + schema_file.write_text(generated, encoding="utf-8") + _generated2, existing2, changed2 = code_to_dbml.code_to_dbml_command( + schema_file=schema_file, + output=tmp_path / "out", + ) + assert existing2 + assert changed2 is False diff --git a/tests/test_full_coverage.py b/tests/test_full_coverage.py deleted file mode 100644 index 66e1f80..0000000 --- a/tests/test_full_coverage.py +++ /dev/null @@ -1,959 +0,0 @@ -import ast -import os -import subprocess -import sys -from pathlib import Path - -import pytest -import typer - -# Add src to path for imports -sys.path.insert(0, str(Path(__file__).parent.parent / "src")) - -from dbml_to_sqlmodel import cli as cli_module -from dbml_to_sqlmodel import generate_app -from dbml_to_sqlmodel.commands import code_to_dbml, generate, info, preview -from dbml_to_sqlmodel.constants import USER_FILE_MARKER -from dbml_to_sqlmodel.core import parser as parser_module -from dbml_to_sqlmodel.core.code_generator import generate_single_model -from dbml_to_sqlmodel.generator import ( - _extract_enums_from_text, - generate_admin_views, - generate_all_files, - generate_enums_file, -) -from dbml_to_sqlmodel.models.file_info import FileInfo, FileStatus -from dbml_to_sqlmodel.models.schema import ColumnInfo, TableInfo -from dbml_to_sqlmodel.sqlmodel_to_dbml import ( - _annotation_to_type, - _canonical_dbml_type, - _columns_equivalent, - _extract_columns, - _format_dbml_column, - _format_table_header, - _format_table_header_like, - _model_files, - _parse_field_kwargs, - _parse_model_file, - _table_name_from_path, - apply_dbml_changes, - apply_dbml_table_updates, - apply_schema_hints, - canonicalize_dbml_text, - generate_dbml_from_models, - normalize_dbml, - normalize_dbml_for_compare, -) -from dbml_to_sqlmodel.utils import diff as diff_module - - -def _write_model(tmp_path: Path, rel_path: str, content: str) -> Path: - file_path = tmp_path / rel_path - file_path.parent.mkdir(parents=True, exist_ok=True) - file_path.write_text(content, encoding="utf-8") - return file_path - - -def _dbml_sample_tables() -> str: - return """ -Table users { - id integer [pk] - name text -} -Table teams { - id integer [pk] - name text -} -Table projects { - id integer [pk] - name text -} -Table notes { - id integer [pk] - body text -} -""" - - -def test_file_info_status_tuple(): - info_obj = FileInfo("a.py", FileStatus.CREATED, True) - assert info_obj.status_tuple == ("created", True) - - -def test_generate_app_main_success(tmp_path, monkeypatch): - dbml_path = tmp_path / "schema.dbml" - dbml_path.write_text( - "Enum status {\n active\n}\nTable users {\n id integer [pk]\n status status\n}\n", - encoding="utf-8", - ) - - output_dir = tmp_path / "out" - monkeypatch.setattr( - sys, - "argv", - ["generate_app.py", str(dbml_path), "-o", str(output_dir)], - ) - - generate_app.main() - - assert (output_dir / "main.py").exists() - assert (output_dir / "models" / "enums.py").exists() - assert (output_dir / "models" / "users" / "model.py").exists() - - -def test_generate_app_main_module_run(tmp_path): - dbml_path = tmp_path / "schema.dbml" - dbml_path.write_text( - "Table users {\n id integer [pk]\n}\n", - encoding="utf-8", - ) - - output_dir = tmp_path / "out" - env = os.environ.copy() - env["PYTHONPATH"] = str(Path(__file__).parent.parent / "src") - subprocess.run( - [ - sys.executable, - "-m", - "dbml_to_sqlmodel.generate_app", - str(dbml_path), - "-o", - str(output_dir), - ], - check=True, - env=env, - stdout=subprocess.PIPE, - stderr=subprocess.PIPE, - text=True, - ) - assert (output_dir / "main.py").exists() - - -def test_generate_app_main_no_tables(tmp_path, monkeypatch): - dbml_path = tmp_path / "schema.dbml" - dbml_path.write_text("", encoding="utf-8") - - output_dir = tmp_path / "out" - monkeypatch.setattr( - sys, - "argv", - ["generate_app.py", str(dbml_path), "-o", str(output_dir)], - ) - - with pytest.raises(SystemExit): - generate_app.main() - - -def test_cli_app_error_paths(monkeypatch): - def raise_keyboard(): - raise KeyboardInterrupt - - monkeypatch.setattr(cli_module, "interactive_menu", raise_keyboard) - cli_module.app() - - def raise_error(): - raise ValueError("boom") - - monkeypatch.setattr(cli_module, "interactive_menu", raise_error) - with pytest.raises(ValueError): - cli_module.app() - - -def test_generate_app_extract_enums_and_admin_views(): - dbml = """ - Enum status { - active - disabled - } - """ - enums = generate_app._extract_enums_from_text(dbml) - assert enums == {"status": ["active", "disabled"]} - - tables = parser_module.parse_dbml("Table users {\n id integer [pk]\n}\n") - auth_code = generate_app.generate_admin_views(tables, admin_auth_enabled=True) - no_auth_code = generate_app.generate_admin_views(tables, admin_auth_enabled=False) - assert "AuthenticationBackend" in auth_code - assert "AuthenticationBackend" not in no_auth_code - - -def test_generate_app_admin_views_labels_and_searchable(): - table = TableInfo( - name="users", - columns=[ - ColumnInfo( - name="name", - type="text", - primary_key=False, - unique=False, - nullable=False, - note="Full name", - ), - ColumnInfo( - name="age", - type="integer", - primary_key=False, - unique=False, - nullable=True, - ), - ], - ) - code = generate_app.generate_admin_views([table], admin_auth_enabled=False) - assert "column_labels" in code - assert "column_searchable_list" in code - - -def test_generator_enums_and_invalid_values(): - enums = _extract_enums_from_text( - """ - enum status { - active, - 1st-place - // comment - } - """ - ) - assert enums == {"status": ["active", "1st-place"]} - - content = generate_enums_file({"status": ["active", "1st-place"]}) - assert "class Status" in content - assert "VALUE_1ST_PLACE" in content - - -def test_generate_all_files_with_enums(): - dbml = """ - Enum status { - active - } - Table users { - id integer [pk] - status status - } - """ - files = generate_all_files(dbml, admin_auth_enabled=False) - assert "models/enums.py" in files - - -def test_generate_single_model_edge_cases(): - enum_table = TableInfo( - name="widgets", - columns=[ - ColumnInfo( - name="id", - type="bool", - primary_key=True, - unique=False, - nullable=False, - note="id note", - ), - ColumnInfo( - name="status", - type="status", - primary_key=True, - unique=False, - nullable=False, - ), - ], - note='note """ triple', - ) - rel_table = TableInfo( - name="posts", - columns=[ - ColumnInfo( - name="user_id", - type="integer", - primary_key=False, - unique=False, - nullable=False, - references=[("users", "id")], - ), - ], - ) - nopk_table = TableInfo( - name="logs", - columns=[ - ColumnInfo( - name="message", - type="text", - primary_key=False, - unique=False, - nullable=True, - ), - ], - ) - all_tables = [enum_table, rel_table, nopk_table] - enums = {"status": ["active"]} - - enum_code = generate_single_model(enum_table, all_tables, enums=enums) - assert '"""note \\"\\"\\" triple"""' in enum_code - assert "DESCRIPTIONS" in enum_code - assert 'description=DESCRIPTIONS["id"]' in enum_code - assert "status: Status" in enum_code - - rel_code = generate_single_model(rel_table, all_tables, enums=enums) - assert "Relationship()" in rel_code - - nopk_code = generate_single_model(nopk_table, all_tables, enums=enums) - assert "class Logs(LogsBase, table=True):\n pass" in nopk_code - - -def test_generate_admin_views_labels_and_searchable(): - table = TableInfo( - name="users", - columns=[ - ColumnInfo( - name="name", - type="varchar", - primary_key=False, - unique=False, - nullable=False, - note="Full name", - ), - ColumnInfo( - name="age", - type="integer", - primary_key=False, - unique=False, - nullable=True, - ), - ], - ) - code = generate_admin_views([table], admin_auth_enabled=True) - assert "column_labels" in code - assert "column_searchable_list" in code - - -def test_generate_all_files_no_tables(): - with pytest.raises(ValueError): - generate_all_files("// empty") - - -def test_extract_dbml_types_and_defaults(): - dbml = """ - // comment - Table users - { - id integer [default: 1] - name varchar(255) [default: "bob"] - active boolean [default: true] - disabled boolean [default: false] - ratio float [default: 1.5] - weird integer [nodefault=3] - // inside comment - invalid_only_name - note: 'ignored' - indexes { - name - } - } - Ref: users.id > other.id - """ - types = parser_module.extract_dbml_types(dbml) - defaults = parser_module.extract_dbml_defaults(dbml) - - assert types["users"]["name"] == "varchar(255)" - assert defaults["users"]["id"] == 1 - assert defaults["users"]["name"] == "bob" - assert defaults["users"]["active"] is True - assert defaults["users"]["disabled"] is False - assert defaults["users"]["ratio"] == 1.5 - - -def test_parse_dbml_enums_from_text(): - dbml = """ - enum status { - active, - disabled // trailing - } - """ - enums = parser_module.parse_dbml_enums(dbml) - assert enums == {"status": ["active", "disabled"]} - - -def test_parse_dbml_enums_missing_name(monkeypatch): - class FakeEnum: - def __init__(self, name, items): - self.name = name - self.items = items - - class FakeItem: - def __init__(self, name): - self.name = name - - class FakeParsed: - enums = [FakeEnum(None, [FakeItem("skip")]), FakeEnum("Status", [FakeItem("active")])] - - monkeypatch.setattr(parser_module, "_parse_dbml_enums_from_text", lambda _text: {}) - monkeypatch.setattr(parser_module, "PyDBML", lambda _text: FakeParsed()) - - enums = parser_module.parse_dbml_enums("enum status { active }") - assert enums == {"Status": ["active"]} - - -def test_parse_dbml_with_fake_pydbml(monkeypatch): - class FakeType: - def __init__(self, name): - self.name = name - - class NoteNoText: - def __init__(self, value): - self.value = value - - def __str__(self): - return self.value - - class FakeColumn: - def __init__( - self, name, col_type, pk=False, unique=False, not_null=False, default=None, note=None - ): - self.name = name - self.type = col_type - self.pk = pk - self.unique = unique - self.not_null = not_null - self.default = default - self.note = note - - class FakeTable: - def __init__(self, name, columns, note=None): - self.name = name - self.columns = columns - self.note = note - - class FakeRef: - def __init__(self, ref_type, col1, col2, table1, table2): - self.type = ref_type - self.col1 = col1 - self.col2 = col2 - self.table1 = table1 - self.table2 = table2 - - users_id = FakeColumn("id", FakeType("int"), pk=True, note=NoteNoText("id note")) - posts_user_id = FakeColumn("user_id", "integer", not_null=True) - - users_table = FakeTable("users", [users_id], note=NoteNoText("users note")) - posts_table = FakeTable("posts", [posts_user_id]) - - ref_forward = FakeRef(">", [posts_user_id], [users_id], posts_table, users_table) - ref_reverse = FakeRef("<", [posts_user_id], [users_id], posts_table, users_table) - - class FakeParsed: - refs = [ref_forward, ref_reverse] - tables = [users_table, posts_table] - - monkeypatch.setattr(parser_module, "PyDBML", lambda _text: FakeParsed()) - - dbml = """ - Table users { - id integer [default: 2] - } - Table posts { - user_id integer - } - """ - tables = parser_module.parse_dbml(dbml) - - users = next(t for t in tables if t.name == "users") - posts = next(t for t in tables if t.name == "posts") - - assert users.note == "users note" - assert users.columns[0].note == "id note" - assert users.columns[0].type == "integer" - assert users.columns[0].default == 2 - assert posts.columns[0].references == [("users", "id")] - assert users.columns[0].references == [("posts", "user_id")] - - -def test_parse_dbml_enums_pydbml_branch(monkeypatch): - dbml = """ - Enum status { - active - disabled - } - """ - - monkeypatch.setattr(parser_module, "_parse_dbml_enums_from_text", lambda _text: {}) - enums = parser_module.parse_dbml_enums(dbml) - - assert enums["status"] == ["active", "disabled"] - - -def test_parse_field_kwargs_and_annotation_to_type(): - source = """ -from typing import Optional -import models -import module - -DESCRIPTIONS = {"name": "User name"} -EXTRA = {"description": "Extra"} -VALUE = "x" -key = "name" -Team = object() - - -def factory(): - return "x" - - -class User: - name: str = Field(description=DESCRIPTIONS["name"], unique=True) - missing: str = Field(description=OTHER["missing"]) - name_key: str = Field(description=DESCRIPTIONS[key]) - optional_ref: Optional[models.Team] = Field(default=None) - simple_opt: Optional[Team] = Field(default=None) - direct_ref: models.Team = Field(default=None) - alias: str = Field(default=VALUE) - attr: str = Field(default=module.VALUE) - extra: str = Field(**EXTRA) - created: str = Field(default_factory=lambda: "x") - called: str = Field(default=factory()) - weird: tuple[int, int] = Field(default=None) -""" - tree = ast.parse(source) - class_node = next(node for node in tree.body if isinstance(node, ast.ClassDef)) - kwargs = [] - first_call = None - for node in class_node.body: - if isinstance(node, ast.AnnAssign) and isinstance(node.value, ast.Call): - if first_call is None: - first_call = node.value - kwargs.append(_parse_field_kwargs(node.value, {"DESCRIPTIONS": {"name": "User name"}})) - - assert any(item.get("description") == "User name" for item in kwargs) - assert any(item.get("unique") is True for item in kwargs) - assert first_call is not None - _parse_field_kwargs(first_call) - - opt_node = next( - node - for node in class_node.body - if isinstance(node, ast.AnnAssign) and node.target.id == "optional_ref" - ) - simple_opt = next( - node - for node in class_node.body - if isinstance(node, ast.AnnAssign) and node.target.id == "simple_opt" - ) - direct_node = next( - node - for node in class_node.body - if isinstance(node, ast.AnnAssign) and node.target.id == "direct_ref" - ) - weird_node = next( - node - for node in class_node.body - if isinstance(node, ast.AnnAssign) and node.target.id == "weird" - ) - - assert _annotation_to_type(opt_node.annotation) == ("Team", True) - assert _annotation_to_type(simple_opt.annotation) == ("Team", True) - assert _annotation_to_type(direct_node.annotation) == ("Team", False) - assert _annotation_to_type(weird_node.annotation) == ("str", False) - - -def test_extract_columns_edge_cases(): - source = """ -from sqlmodel import Field, Relationship - -class Model: - id: int - raw: int = 1 - rel: "Other" = Relationship() - skip: int = Other() - ok: int = Field(default=1) -""" - tree = ast.parse(source) - class_node = next(node for node in tree.body if isinstance(node, ast.ClassDef)) - - columns = _extract_columns(class_node, None) - names = {col.name for col in columns} - assert "id" in names - assert "ok" in names - assert "raw" not in names - assert "rel" not in names - assert "skip" not in names - - -def test_parse_model_file_missing_table_class(tmp_path): - model_path = _write_model(tmp_path, "models/user/model.py", "class Base: pass") - with pytest.raises(ValueError): - _parse_model_file(model_path, {}) - - -def test_parse_model_file_duplicate_columns(tmp_path): - model_py = """ -from sqlmodel import Field, SQLModel - -class Base(SQLModel): - id: int = Field(primary_key=True) - -class User(Base, table=True): - id: int = Field(primary_key=True) -""" - model_path = _write_model(tmp_path, "models/user/model.py", model_py) - table = _parse_model_file(model_path, {}) - assert len([c for c in table.columns if c.name == "id"]) == 1 - - -def test_sqlmodel_to_dbml_generate_and_apply(tmp_path): - models_dir = tmp_path / "models" - models_dir.mkdir() - - enums_py = """ -from enum import Enum - -class Status(Enum): - __dbml_name__ = "status" - ACTIVE = "active" - -class NotEnum: - pass -""" - _write_model(tmp_path, "models/enums.py", enums_py) - - model_py = """ -from typing import Optional -import models -from sqlmodel import Field, SQLModel, Relationship - -DESCRIPTIONS = {"name": "User name"} -EXTRA = {"description": "Extra"} - - -def factory(): - return "x" - - -class Base(SQLModel): - id: int = Field(primary_key=True) - - -class User(Base, table=True): - '''User table.''' - name: str = Field(description=DESCRIPTIONS["name"], unique=True) - other: str = Field(description=OTHER["missing"]) - status: Status = Field(default="active") - team_id: Optional[models.Team] = Field(default=None, foreign_key="teams.id") - extra: str = Field(**EXTRA) - created: str = Field(default_factory=lambda: "x") - called: str = Field(default=factory()) - rel: list["Team"] = Relationship(back_populates="users") -""" - _write_model(tmp_path, "models/user/model.py", model_py) - teams_py = """ -from sqlmodel import Field, SQLModel - -class Team(SQLModel, table=True): - id: int = Field(primary_key=True) -""" - _write_model(tmp_path, "models/teams/model.py", teams_py) - - dbml = generate_dbml_from_models(models_dir) - assert "Table user" in dbml - assert "status status" in dbml - assert "ref: > teams.id" in dbml - assert 'note: "User name"' in dbml - - schema_dbml = """ - Table teams { - id integer [pk] - } - Table user { - id integer [pk, default: 10] - name text - status text - team_id integer - } - """ - hinted = apply_schema_hints(dbml, schema_dbml) - assert "default: 10" in hinted - assert "status text" in hinted - - -def test_sqlmodel_to_dbml_helpers_and_changes(): - col = ColumnInfo( - name="name", - type="text", - primary_key=False, - unique=True, - nullable=False, - default="bob", - references=[("teams", "id")], - note='note "quoted"', - ) - formatted = _format_dbml_column(col) - assert 'default: "bob"' in formatted - assert "ref: > teams.id" in formatted - assert 'note: "note \\"quoted\\""' in formatted - - header = _format_table_header_like("table users {", TableInfo("users", [], note="hi")) - header_caps = _format_table_header_like("Table users {", TableInfo("users", [])) - header_full = _format_table_header(TableInfo("users", [], note="hi")) - header_plain = _format_table_header(TableInfo("users", [])) - assert header.startswith("table users") - assert header_caps == "Table users {" - assert "note:" in header_full - assert header_plain == "Table users {" - - assert _canonical_dbml_type("serial4") == "integer" - assert _canonical_dbml_type("CUSTOM") == "custom" - assert _columns_equivalent(col, col) is True - - plain_col = ColumnInfo( - name="id", - type="integer", - primary_key=False, - unique=False, - nullable=True, - ) - assert _format_dbml_column(plain_col) == " id integer" - num_default = ColumnInfo( - name="count", - type="integer", - primary_key=False, - unique=False, - nullable=True, - default=5, - ) - assert "default: 5" in _format_dbml_column(num_default) - - existing = """ -Table legacy { - id integer [primary key] -} - -Table users [note: \"Old note\"] { - id integer [primary key] - name text - // keep -} -""" - generated = """ -Table users [note: \"New note\"] { - id integer [primary key] - name text [not null] -} - -Table teams { - id integer [primary key] -} -""" - updated = apply_dbml_changes(existing, generated) - assert "New note" in updated - assert "// keep" in updated - assert "Table teams" in updated - - updated_tables = apply_dbml_table_updates(existing, generated) - assert "not null" in updated_tables - - canonical = canonicalize_dbml_text("Table users {\n id serial4\n}\n") - assert "id integer" in canonical - assert "Note: keep" in canonicalize_dbml_text("Note: keep\n") - - assert normalize_dbml("Table users {\n id integer [pk]\n}\n") - assert normalize_dbml_for_compare("Table users {\n id serial4 [pk]\n}\n") - - -def test_columns_equivalent_false_cases(): - base = ColumnInfo( - name="id", - type="integer", - primary_key=False, - unique=False, - nullable=True, - default=None, - references=None, - note=None, - ) - assert _columns_equivalent(base, ColumnInfo(**{**base.__dict__, "type": "text"})) is False - assert _columns_equivalent(base, ColumnInfo(**{**base.__dict__, "primary_key": True})) is False - assert _columns_equivalent(base, ColumnInfo(**{**base.__dict__, "unique": True})) is False - assert _columns_equivalent(base, ColumnInfo(**{**base.__dict__, "nullable": False})) is False - assert _columns_equivalent(base, ColumnInfo(**{**base.__dict__, "default": 1})) is False - assert ( - _columns_equivalent(base, ColumnInfo(**{**base.__dict__, "references": [("t", "id")]})) - is False - ) - assert _columns_equivalent(base, ColumnInfo(**{**base.__dict__, "note": "x"})) is False - - -def test_apply_dbml_table_updates_unchanged_block(): - dbml = """ - Table users { - id integer [pk] - } - """ - updated = apply_dbml_table_updates(dbml, dbml) - assert "Table users" in updated - - -def test_sqlmodel_to_dbml_model_files_helpers(tmp_path): - models_dir = tmp_path / "models" - assert list(_model_files(models_dir)) == [] - - models_dir.mkdir() - file_path = _write_model(tmp_path, "models/users/model.py", "class User: ...") - assert list(_model_files(models_dir)) == [file_path] - assert _table_name_from_path(file_path) == "users" - - -def test_generate_dbml_from_models_no_files(tmp_path): - with pytest.raises(ValueError): - generate_dbml_from_models(tmp_path / "models") - - -def test_commands_preview_info_generate(tmp_path): - schema_file = tmp_path / "schema.dbml" - schema_file.write_text(_dbml_sample_tables(), encoding="utf-8") - - output = tmp_path / "out" - output.mkdir() - - generated_files = generate_all_files(schema_file.read_text(encoding="utf-8")) - - for rel_path, content in generated_files.items(): - if not rel_path.endswith("/model.py"): - continue - file_path = output / rel_path - file_path.parent.mkdir(parents=True, exist_ok=True) - if "users" in rel_path: - file_path.write_text(content, encoding="utf-8") - elif "teams" in rel_path: - file_path.write_text(content + "\nEXTRA = 1\n", encoding="utf-8") - elif "projects" in rel_path: - file_path.write_text(USER_FILE_MARKER + "\n" + content, encoding="utf-8") - else: - pass - - summary = preview.preview_command( - schema_file=schema_file, output=output, show_all=True, show_new=True - ) - assert summary["new"] >= 1 - assert summary["modified"] >= 1 - assert summary["protected"] >= 1 - - preview.preview_command(schema_file=schema_file, output=output, show_all=False, show_new=False) - - info.info_command(schema_file=schema_file, output=output) - - generate.generate_command( - schema_file=schema_file, output=output, force=False, admin_auth_enabled=False - ) - - generate.generate_command( - schema_file=schema_file, output=output, force=True, admin_auth_enabled=True - ) - - -def test_commands_generate_all_files_errors(tmp_path, monkeypatch): - schema_file = tmp_path / "schema.dbml" - schema_file.write_text("Table users {\n id integer [pk]\n}\n", encoding="utf-8") - - def raise_generate(*_args, **_kwargs): - raise RuntimeError("boom") - - monkeypatch.setattr(preview, "generate_all_files", raise_generate) - with pytest.raises(typer.Exit): - preview.preview_command(schema_file=schema_file, output=tmp_path) - - monkeypatch.setattr(info, "generate_all_files", raise_generate) - with pytest.raises(typer.Exit): - info.info_command(schema_file=schema_file, output=tmp_path) - - monkeypatch.setattr(generate, "generate_all_files", raise_generate) - with pytest.raises(typer.Exit): - generate.generate_command(schema_file=schema_file, output=tmp_path) - - -def test_commands_error_paths(tmp_path, monkeypatch): - schema_file = tmp_path / "schema.dbml" - schema_file.write_text("Table users { id integer [pk] }", encoding="utf-8") - - def bad_read(*_args, **_kwargs): - raise OSError("boom") - - monkeypatch.setattr(Path, "read_text", bad_read) - - with pytest.raises(typer.Exit): - preview.preview_command(schema_file=schema_file) - - with pytest.raises(typer.Exit): - info.info_command(schema_file=schema_file) - - with pytest.raises(typer.Exit): - generate.generate_command(schema_file=schema_file) - - -def test_code_to_dbml_command_branches(tmp_path): - schema_file = tmp_path / "schema.dbml" - - with pytest.raises(Exception): - code_to_dbml.code_to_dbml_command(schema_file=schema_file, output=tmp_path) - - models_dir = tmp_path / "out" / "models" - models_dir.mkdir(parents=True) - _write_model(tmp_path, "out/models/enums.py", "from enum import Enum\n") - model_py = """ -from sqlmodel import Field, SQLModel - -class User(SQLModel, table=True): - id: int = Field(primary_key=True) -""" - _write_model(tmp_path, "out/models/user/model.py", model_py) - - generated, existing, changed = code_to_dbml.code_to_dbml_command( - schema_file=schema_file, - output=tmp_path / "out", - ) - assert existing == "" - assert changed is True - assert "Table user" in generated - - schema_file.write_text(generated, encoding="utf-8") - _generated2, existing2, changed2 = code_to_dbml.code_to_dbml_command( - schema_file=schema_file, - output=tmp_path / "out", - ) - assert existing2 - assert changed2 is False - - -def test_diff_helpers(tmp_path, monkeypatch): - diff_module.print_diff("", "file.txt") - diff_module.print_diff("-a\n+b", "file.txt") - - file_path = tmp_path / "file.txt" - assert diff_module.apply_diff_to_file(file_path, "old", "new") is True - assert file_path.read_text(encoding="utf-8") == "new" - - assert diff_module.apply_diff_to_file(file_path, "new", "new") is False - - original_generate_diff = diff_module.generate_diff - monkeypatch.setattr(diff_module, "generate_diff", lambda *_args, **_kwargs: "") - file_path.write_text("old", encoding="utf-8") - assert diff_module.apply_diff_to_file(file_path, "old", "new") is False - - monkeypatch.setattr(diff_module, "generate_diff", original_generate_diff) - - file_path.write_text("a\nb\nc\n", encoding="utf-8") - assert diff_module.apply_diff_to_file(file_path, "a\nb\nc\n", "a\nx\nc\n") is True - - file_path.write_text("z\nb\ny\n", encoding="utf-8") - assert diff_module.apply_diff_to_file(file_path, "a\nb\nc\n", "a\nx\nc\n") is True - - file_path.write_text("a\nb\nz\nc\nd\n", encoding="utf-8") - assert diff_module.apply_diff_to_file(file_path, "a\nb\nc\nd\n", "a\nx\ny\nd\n") is True - - file_path.write_text("q\nw\n", encoding="utf-8") - assert diff_module.apply_diff_to_file(file_path, "a\nb\n", "a\nx\n") is False - - file_path.write_text("", encoding="utf-8") - assert diff_module.apply_diff_to_file(file_path, "", "a\n") is False - - class Boom(Exception): - pass - - def raise_patch(_text): - raise Boom("fail") - - monkeypatch.setattr(diff_module, "PatchSet", raise_patch) - assert diff_module.apply_diff_to_file(file_path, "a\n", "b\n") is False diff --git a/tests/test_generated_app_smoke.py b/tests/test_generated_app_smoke.py new file mode 100644 index 0000000..39e07de --- /dev/null +++ b/tests/test_generated_app_smoke.py @@ -0,0 +1,91 @@ +"""Smoke tests that exercise the *generated* application output. + +The other test modules assert on the generator's string output. These tests +guard against regressions where the generator emits code that is broken in +practice (syntax errors, deprecated APIs, wrong types, unsatisfiable +requirements) even though those string assertions still pass. + +Only the generator itself is exercised — the generated app's runtime +dependencies (fastapi, sqlmodel, ...) are *not* required to run these tests. +""" + +import compileall +import sys +from pathlib import Path + +# Add src to path for imports +sys.path.insert(0, str(Path(__file__).parent.parent / "src")) + +from dbml_to_sqlmodel.generator import generate_all_files + +SCHEMA = """ +Enum post_status { + draft + published +} + +Table users { + id serial [pk] + email text [not null, unique] + created_at timestamp [not null] +} + +Table post_tags { + id serial [pk] + user_id integer [ref: > users.id] + status post_status + tagged_on date +} +""" + + +def _write_files(files: dict[str, str], root: Path) -> None: + for rel_path, content in files.items(): + path = root / rel_path + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(content, encoding="utf-8") + + +def test_generated_files_are_valid_python(tmp_path): + """Every generated .py file must at least compile.""" + files = generate_all_files(SCHEMA, admin_auth_enabled=True) + _write_files(files, tmp_path) + # compile_dir returns True only if every file compiled without errors. + assert compileall.compile_dir(str(tmp_path), quiet=1) + + +def test_generated_main_uses_env_and_lifespan(): + """main.py must read DATABASE_URL from the env and use the lifespan API.""" + main = generate_all_files(SCHEMA)["main.py"] + assert 'os.getenv("DATABASE_URL"' in main + assert "lifespan=lifespan" in main + # Deprecated FastAPI startup hook must not be emitted. + assert "on_event" not in main + + +def test_generated_models_map_temporal_types(): + """Temporal DBML types must become datetime/date, not str.""" + files = generate_all_files(SCHEMA) + + user_model = files["models/users/model.py"] + assert "from datetime import datetime" in user_model + assert "created_at: datetime" in user_model + + post_model = files["models/post_tags/model.py"] + assert "from datetime import date" in post_model + # Multi-word table names become proper PascalCase class names. + assert "class PostTags(" in post_model + assert "class Post_tags(" not in post_model + + +def test_generated_requirements_are_satisfiable(): + """requirements.txt must not pin versions that do not exist.""" + reqs = generate_all_files(SCHEMA)["requirements.txt"] + for package in ("fastapi", "sqlmodel", "sqladmin", "fastcrud", "aiosqlite"): + assert package in reqs + # Regression guard: sqladmin never had a 1.x release. + assert "sqladmin>=1." not in reqs + assert "sqladmin>=0.22.0" in reqs + # The async SQLAlchemy engine needs greenlet, which is not auto-installed + # on every platform/Python combination (e.g. Python 3.13 on macOS arm64). + assert "greenlet" in reqs diff --git a/tests/test_generator.py b/tests/test_generator.py index e114794..59000a7 100644 --- a/tests/test_generator.py +++ b/tests/test_generator.py @@ -1,17 +1,22 @@ import sys from pathlib import Path +import pytest + # Add src to path for imports sys.path.insert(0, str(Path(__file__).parent.parent / "src")) +from dbml_to_sqlmodel.core import parser as parser_module from dbml_to_sqlmodel.core.parser import parse_dbml from dbml_to_sqlmodel.generator import ( generate_admin_views, generate_all_files, + generate_enums_file, generate_main_app, generate_requirements, ) from dbml_to_sqlmodel.models.config_models import AppConfig +from dbml_to_sqlmodel.models.schema import ColumnInfo, TableInfo DBML_SAMPLE = """ Table users { @@ -132,3 +137,108 @@ def test_generate_all_files_multiple_tables(): # Should have model files for both tables assert any("users" in path for path in files.keys()) assert any("posts" in path for path in files.keys()) + + +def test_extract_enums_and_admin_views(): + dbml = """ + Enum status { + active + disabled + } + """ + enums = parser_module.parse_dbml_enums(dbml) + assert enums == {"status": ["active", "disabled"]} + + tables = parser_module.parse_dbml("Table users {\n id integer [pk]\n}\n") + auth_code = generate_admin_views(tables, admin_auth_enabled=True) + no_auth_code = generate_admin_views(tables, admin_auth_enabled=False) + assert "AuthenticationBackend" in auth_code + assert "AuthenticationBackend" not in no_auth_code + + +def test_admin_views_labels_and_searchable(): + table = TableInfo( + name="users", + columns=[ + ColumnInfo( + name="name", + type="text", + primary_key=False, + unique=False, + nullable=False, + note="Full name", + ), + ColumnInfo( + name="age", + type="integer", + primary_key=False, + unique=False, + nullable=True, + ), + ], + ) + code = generate_admin_views([table], admin_auth_enabled=False) + assert "column_labels" in code + assert "column_searchable_list" in code + + +def test_generate_admin_views_labels_and_searchable(): + table = TableInfo( + name="users", + columns=[ + ColumnInfo( + name="name", + type="varchar", + primary_key=False, + unique=False, + nullable=False, + note="Full name", + ), + ColumnInfo( + name="age", + type="integer", + primary_key=False, + unique=False, + nullable=True, + ), + ], + ) + code = generate_admin_views([table], admin_auth_enabled=True) + assert "column_labels" in code + assert "column_searchable_list" in code + + +def test_generator_enums_and_invalid_values(): + enums = parser_module.parse_dbml_enums( + """ + enum status { + active, + 1st-place + // comment + } + """ + ) + assert enums == {"status": ["active", "1st-place"]} + + content = generate_enums_file({"status": ["active", "1st-place"]}) + assert "class Status" in content + assert "VALUE_1ST_PLACE" in content + + +def test_generate_all_files_with_enums(): + dbml = """ + Enum status { + active + } + Table users { + id integer [pk] + status status + } + """ + files = generate_all_files(dbml, admin_auth_enabled=False) + assert "models/enums.py" in files + + +def test_generate_all_files_no_tables(): + with pytest.raises(ValueError): + generate_all_files("// empty") diff --git a/tests/test_parser.py b/tests/test_parser.py index 8076849..5616556 100644 --- a/tests/test_parser.py +++ b/tests/test_parser.py @@ -4,6 +4,7 @@ # Add src to path for imports sys.path.insert(0, str(Path(__file__).parent.parent / "src")) +from dbml_to_sqlmodel.core import parser as parser_module from dbml_to_sqlmodel.core.parser import parse_dbml @@ -249,3 +250,153 @@ def test_parse_bidirectional_relationship(): assert len(author_refs) > 0 assert len(editor_refs) > 0 + + +def test_extract_dbml_types_and_defaults(): + dbml = """ + // comment + Table users + { + id integer [default: 1] + name varchar(255) [default: "bob"] + active boolean [default: true] + disabled boolean [default: false] + ratio float [default: 1.5] + weird integer [nodefault=3] + // inside comment + invalid_only_name + note: 'ignored' + indexes { + name + } + } + Ref: users.id > other.id + """ + types = parser_module.extract_dbml_types(dbml) + defaults = parser_module.extract_dbml_defaults(dbml) + + assert types["users"]["name"] == "varchar(255)" + assert defaults["users"]["id"] == 1 + assert defaults["users"]["name"] == "bob" + assert defaults["users"]["active"] is True + assert defaults["users"]["disabled"] is False + assert defaults["users"]["ratio"] == 1.5 + + +def test_parse_dbml_enums_from_text(): + dbml = """ + enum status { + active, + disabled // trailing + } + """ + enums = parser_module.parse_dbml_enums(dbml) + assert enums == {"status": ["active", "disabled"]} + + +def test_parse_dbml_enums_missing_name(monkeypatch): + class FakeEnum: + def __init__(self, name, items): + self.name = name + self.items = items + + class FakeItem: + def __init__(self, name): + self.name = name + + class FakeParsed: + enums = [FakeEnum(None, [FakeItem("skip")]), FakeEnum("Status", [FakeItem("active")])] + + monkeypatch.setattr(parser_module, "_parse_dbml_enums_from_text", lambda _text: {}) + monkeypatch.setattr(parser_module, "PyDBML", lambda _text: FakeParsed()) + + enums = parser_module.parse_dbml_enums("enum status { active }") + assert enums == {"Status": ["active"]} + + +def test_parse_dbml_with_fake_pydbml(monkeypatch): + class FakeType: + def __init__(self, name): + self.name = name + + class NoteNoText: + def __init__(self, value): + self.value = value + + def __str__(self): + return self.value + + class FakeColumn: + def __init__( + self, name, col_type, pk=False, unique=False, not_null=False, default=None, note=None + ): + self.name = name + self.type = col_type + self.pk = pk + self.unique = unique + self.not_null = not_null + self.default = default + self.note = note + + class FakeTable: + def __init__(self, name, columns, note=None): + self.name = name + self.columns = columns + self.note = note + + class FakeRef: + def __init__(self, ref_type, col1, col2, table1, table2): + self.type = ref_type + self.col1 = col1 + self.col2 = col2 + self.table1 = table1 + self.table2 = table2 + + users_id = FakeColumn("id", FakeType("int"), pk=True, note=NoteNoText("id note")) + posts_user_id = FakeColumn("user_id", "integer", not_null=True) + + users_table = FakeTable("users", [users_id], note=NoteNoText("users note")) + posts_table = FakeTable("posts", [posts_user_id]) + + ref_forward = FakeRef(">", [posts_user_id], [users_id], posts_table, users_table) + ref_reverse = FakeRef("<", [posts_user_id], [users_id], posts_table, users_table) + + class FakeParsed: + refs = [ref_forward, ref_reverse] + tables = [users_table, posts_table] + + monkeypatch.setattr(parser_module, "PyDBML", lambda _text: FakeParsed()) + + dbml = """ + Table users { + id integer [default: 2] + } + Table posts { + user_id integer + } + """ + tables = parser_module.parse_dbml(dbml) + + users = next(t for t in tables if t.name == "users") + posts = next(t for t in tables if t.name == "posts") + + assert users.note == "users note" + assert users.columns[0].note == "id note" + assert users.columns[0].type == "integer" + assert users.columns[0].default == 2 + assert posts.columns[0].references == [("users", "id")] + assert users.columns[0].references == [("posts", "user_id")] + + +def test_parse_dbml_enums_pydbml_branch(monkeypatch): + dbml = """ + Enum status { + active + disabled + } + """ + + monkeypatch.setattr(parser_module, "_parse_dbml_enums_from_text", lambda _text: {}) + enums = parser_module.parse_dbml_enums(dbml) + + assert enums["status"] == ["active", "disabled"] diff --git a/tests/test_sqlmodel_to_dbml.py b/tests/test_sqlmodel_to_dbml.py new file mode 100644 index 0000000..7efd01d --- /dev/null +++ b/tests/test_sqlmodel_to_dbml.py @@ -0,0 +1,363 @@ +"""Tests for the reverse conversion (code -> DBML) module.""" + +import ast +import sys +from pathlib import Path + +import pytest + +# Add src to path for imports +sys.path.insert(0, str(Path(__file__).parent.parent / "src")) + +from dbml_to_sqlmodel.models.schema import ColumnInfo, TableInfo +from dbml_to_sqlmodel.sqlmodel_to_dbml import ( + _annotation_to_type, + _canonical_dbml_type, + _columns_equivalent, + _extract_columns, + _format_dbml_column, + _format_table_header, + _format_table_header_like, + _model_files, + _parse_field_kwargs, + _parse_model_file, + _table_name_from_path, + apply_dbml_changes, + apply_dbml_table_updates, + apply_schema_hints, + canonicalize_dbml_text, + generate_dbml_from_models, + normalize_dbml, + normalize_dbml_for_compare, +) + + +def _write_model(tmp_path: Path, rel_path: str, content: str) -> Path: + file_path = tmp_path / rel_path + file_path.parent.mkdir(parents=True, exist_ok=True) + file_path.write_text(content, encoding="utf-8") + return file_path + + +def test_parse_field_kwargs_and_annotation_to_type(): + source = """ +from typing import Optional +import models +import module + +DESCRIPTIONS = {"name": "User name"} +EXTRA = {"description": "Extra"} +VALUE = "x" +key = "name" +Team = object() + + +def factory(): + return "x" + + +class User: + name: str = Field(description=DESCRIPTIONS["name"], unique=True) + missing: str = Field(description=OTHER["missing"]) + name_key: str = Field(description=DESCRIPTIONS[key]) + optional_ref: Optional[models.Team] = Field(default=None) + simple_opt: Optional[Team] = Field(default=None) + direct_ref: models.Team = Field(default=None) + alias: str = Field(default=VALUE) + attr: str = Field(default=module.VALUE) + extra: str = Field(**EXTRA) + created: str = Field(default_factory=lambda: "x") + called: str = Field(default=factory()) + weird: tuple[int, int] = Field(default=None) +""" + tree = ast.parse(source) + class_node = next(node for node in tree.body if isinstance(node, ast.ClassDef)) + kwargs = [] + first_call = None + for node in class_node.body: + if isinstance(node, ast.AnnAssign) and isinstance(node.value, ast.Call): + if first_call is None: + first_call = node.value + kwargs.append(_parse_field_kwargs(node.value, {"DESCRIPTIONS": {"name": "User name"}})) + + assert any(item.get("description") == "User name" for item in kwargs) + assert any(item.get("unique") is True for item in kwargs) + assert first_call is not None + _parse_field_kwargs(first_call) + + opt_node = next( + node + for node in class_node.body + if isinstance(node, ast.AnnAssign) and node.target.id == "optional_ref" + ) + simple_opt = next( + node + for node in class_node.body + if isinstance(node, ast.AnnAssign) and node.target.id == "simple_opt" + ) + direct_node = next( + node + for node in class_node.body + if isinstance(node, ast.AnnAssign) and node.target.id == "direct_ref" + ) + weird_node = next( + node + for node in class_node.body + if isinstance(node, ast.AnnAssign) and node.target.id == "weird" + ) + + assert _annotation_to_type(opt_node.annotation) == ("Team", True) + assert _annotation_to_type(simple_opt.annotation) == ("Team", True) + assert _annotation_to_type(direct_node.annotation) == ("Team", False) + assert _annotation_to_type(weird_node.annotation) == ("str", False) + + +def test_extract_columns_edge_cases(): + source = """ +from sqlmodel import Field, Relationship + +class Model: + id: int + raw: int = 1 + rel: "Other" = Relationship() + skip: int = Other() + ok: int = Field(default=1) +""" + tree = ast.parse(source) + class_node = next(node for node in tree.body if isinstance(node, ast.ClassDef)) + + columns = _extract_columns(class_node, None) + names = {col.name for col in columns} + assert "id" in names + assert "ok" in names + assert "raw" not in names + assert "rel" not in names + assert "skip" not in names + + +def test_parse_model_file_missing_table_class(tmp_path): + model_path = _write_model(tmp_path, "models/user/model.py", "class Base: pass") + with pytest.raises(ValueError): + _parse_model_file(model_path, {}) + + +def test_parse_model_file_duplicate_columns(tmp_path): + model_py = """ +from sqlmodel import Field, SQLModel + +class Base(SQLModel): + id: int = Field(primary_key=True) + +class User(Base, table=True): + id: int = Field(primary_key=True) +""" + model_path = _write_model(tmp_path, "models/user/model.py", model_py) + table = _parse_model_file(model_path, {}) + assert len([c for c in table.columns if c.name == "id"]) == 1 + + +def test_sqlmodel_to_dbml_generate_and_apply(tmp_path): + models_dir = tmp_path / "models" + models_dir.mkdir() + + enums_py = """ +from enum import Enum + +class Status(Enum): + __dbml_name__ = "status" + ACTIVE = "active" + +class NotEnum: + pass +""" + _write_model(tmp_path, "models/enums.py", enums_py) + + model_py = """ +from typing import Optional +import models +from sqlmodel import Field, SQLModel, Relationship + +DESCRIPTIONS = {"name": "User name"} +EXTRA = {"description": "Extra"} + + +def factory(): + return "x" + + +class Base(SQLModel): + id: int = Field(primary_key=True) + + +class User(Base, table=True): + '''User table.''' + name: str = Field(description=DESCRIPTIONS["name"], unique=True) + other: str = Field(description=OTHER["missing"]) + status: Status = Field(default="active") + team_id: Optional[models.Team] = Field(default=None, foreign_key="teams.id") + extra: str = Field(**EXTRA) + created: str = Field(default_factory=lambda: "x") + called: str = Field(default=factory()) + rel: list["Team"] = Relationship(back_populates="users") +""" + _write_model(tmp_path, "models/user/model.py", model_py) + teams_py = """ +from sqlmodel import Field, SQLModel + +class Team(SQLModel, table=True): + id: int = Field(primary_key=True) +""" + _write_model(tmp_path, "models/teams/model.py", teams_py) + + dbml = generate_dbml_from_models(models_dir) + assert "Table user" in dbml + assert "status status" in dbml + assert "ref: > teams.id" in dbml + assert 'note: "User name"' in dbml + + schema_dbml = """ + Table teams { + id integer [pk] + } + Table user { + id integer [pk, default: 10] + name text + status text + team_id integer + } + """ + hinted = apply_schema_hints(dbml, schema_dbml) + assert "default: 10" in hinted + assert "status text" in hinted + + +def test_sqlmodel_to_dbml_helpers_and_changes(): + col = ColumnInfo( + name="name", + type="text", + primary_key=False, + unique=True, + nullable=False, + default="bob", + references=[("teams", "id")], + note='note "quoted"', + ) + formatted = _format_dbml_column(col) + assert 'default: "bob"' in formatted + assert "ref: > teams.id" in formatted + assert 'note: "note \\"quoted\\""' in formatted + + header = _format_table_header_like("table users {", TableInfo("users", [], note="hi")) + header_caps = _format_table_header_like("Table users {", TableInfo("users", [])) + header_full = _format_table_header(TableInfo("users", [], note="hi")) + header_plain = _format_table_header(TableInfo("users", [])) + assert header.startswith("table users") + assert header_caps == "Table users {" + assert "note:" in header_full + assert header_plain == "Table users {" + + assert _canonical_dbml_type("serial4") == "integer" + assert _canonical_dbml_type("CUSTOM") == "custom" + assert _columns_equivalent(col, col) is True + + plain_col = ColumnInfo( + name="id", + type="integer", + primary_key=False, + unique=False, + nullable=True, + ) + assert _format_dbml_column(plain_col) == " id integer" + num_default = ColumnInfo( + name="count", + type="integer", + primary_key=False, + unique=False, + nullable=True, + default=5, + ) + assert "default: 5" in _format_dbml_column(num_default) + + existing = """ +Table legacy { + id integer [primary key] +} + +Table users [note: \"Old note\"] { + id integer [primary key] + name text + // keep +} +""" + generated = """ +Table users [note: \"New note\"] { + id integer [primary key] + name text [not null] +} + +Table teams { + id integer [primary key] +} +""" + updated = apply_dbml_changes(existing, generated) + assert "New note" in updated + assert "// keep" in updated + assert "Table teams" in updated + + updated_tables = apply_dbml_table_updates(existing, generated) + assert "not null" in updated_tables + + canonical = canonicalize_dbml_text("Table users {\n id serial4\n}\n") + assert "id integer" in canonical + assert "Note: keep" in canonicalize_dbml_text("Note: keep\n") + + assert normalize_dbml("Table users {\n id integer [pk]\n}\n") + assert normalize_dbml_for_compare("Table users {\n id serial4 [pk]\n}\n") + + +def test_columns_equivalent_false_cases(): + base = ColumnInfo( + name="id", + type="integer", + primary_key=False, + unique=False, + nullable=True, + default=None, + references=None, + note=None, + ) + assert _columns_equivalent(base, ColumnInfo(**{**base.__dict__, "type": "text"})) is False + assert _columns_equivalent(base, ColumnInfo(**{**base.__dict__, "primary_key": True})) is False + assert _columns_equivalent(base, ColumnInfo(**{**base.__dict__, "unique": True})) is False + assert _columns_equivalent(base, ColumnInfo(**{**base.__dict__, "nullable": False})) is False + assert _columns_equivalent(base, ColumnInfo(**{**base.__dict__, "default": 1})) is False + assert ( + _columns_equivalent(base, ColumnInfo(**{**base.__dict__, "references": [("t", "id")]})) + is False + ) + assert _columns_equivalent(base, ColumnInfo(**{**base.__dict__, "note": "x"})) is False + + +def test_apply_dbml_table_updates_unchanged_block(): + dbml = """ + Table users { + id integer [pk] + } + """ + updated = apply_dbml_table_updates(dbml, dbml) + assert "Table users" in updated + + +def test_sqlmodel_to_dbml_model_files_helpers(tmp_path): + models_dir = tmp_path / "models" + assert list(_model_files(models_dir)) == [] + + models_dir.mkdir() + file_path = _write_model(tmp_path, "models/users/model.py", "class User: ...") + assert list(_model_files(models_dir)) == [file_path] + assert _table_name_from_path(file_path) == "users" + + +def test_generate_dbml_from_models_no_files(tmp_path): + with pytest.raises(ValueError): + generate_dbml_from_models(tmp_path / "models") diff --git a/tests/test_style.py b/tests/test_style.py deleted file mode 100644 index b25329d..0000000 --- a/tests/test_style.py +++ /dev/null @@ -1,37 +0,0 @@ -#!/usr/bin/env python3 -"""Test script to check questionary styling.""" - -import questionary -from questionary import Style - -# Custom style for testing -test_style = Style( - [ - ("qmark", "fg:#00d7ff bold"), - ("question", "fg:#ffffff bold"), - ("answer", "fg:#00ff87 bold"), - ("pointer", "fg:#ffff00 bold"), - ("highlighted", "fg:#000000 bold bg:#ffff00 underline"), - ("selected", "fg:#00ff87"), - ("text", ""), - ] -) - - -def main(): - choice = questionary.select( - "Test menu - move with arrows to see highlighting:", - choices=[ - "Option 1", - "Option 2 (with underline on yellow bg)", - "Option 3", - "Exit", - ], - style=test_style, - ).ask() - - print(f"\nYou selected: {choice}") - - -if __name__ == "__main__": - main() diff --git a/tests/test_utils.py b/tests/test_utils.py index ee6362f..19f513c 100644 --- a/tests/test_utils.py +++ b/tests/test_utils.py @@ -9,6 +9,8 @@ sys.path.insert(0, str(Path(__file__).parent.parent / "src")) from dbml_to_sqlmodel.constants import USER_FILE_MARKER +from dbml_to_sqlmodel.models.file_info import FileInfo, FileStatus +from dbml_to_sqlmodel.utils import diff as diff_module from dbml_to_sqlmodel.utils.diff import ( apply_diff_to_file, generate_diff, @@ -346,3 +348,50 @@ def test_print_file_status_table_multiple_files(self): "unchanged.py": ("unchanged", False), } print_file_status_table(files_status) + + +def test_file_info_status_tuple(): + info_obj = FileInfo("a.py", FileStatus.CREATED, True) + assert info_obj.status_tuple == ("created", True) + + +def test_diff_helpers(tmp_path, monkeypatch): + diff_module.print_diff("", "file.txt") + diff_module.print_diff("-a\n+b", "file.txt") + + file_path = tmp_path / "file.txt" + assert diff_module.apply_diff_to_file(file_path, "old", "new") is True + assert file_path.read_text(encoding="utf-8") == "new" + + assert diff_module.apply_diff_to_file(file_path, "new", "new") is False + + original_generate_diff = diff_module.generate_diff + monkeypatch.setattr(diff_module, "generate_diff", lambda *_args, **_kwargs: "") + file_path.write_text("old", encoding="utf-8") + assert diff_module.apply_diff_to_file(file_path, "old", "new") is False + + monkeypatch.setattr(diff_module, "generate_diff", original_generate_diff) + + file_path.write_text("a\nb\nc\n", encoding="utf-8") + assert diff_module.apply_diff_to_file(file_path, "a\nb\nc\n", "a\nx\nc\n") is True + + file_path.write_text("z\nb\ny\n", encoding="utf-8") + assert diff_module.apply_diff_to_file(file_path, "a\nb\nc\n", "a\nx\nc\n") is True + + file_path.write_text("a\nb\nz\nc\nd\n", encoding="utf-8") + assert diff_module.apply_diff_to_file(file_path, "a\nb\nc\nd\n", "a\nx\ny\nd\n") is True + + file_path.write_text("q\nw\n", encoding="utf-8") + assert diff_module.apply_diff_to_file(file_path, "a\nb\n", "a\nx\n") is False + + file_path.write_text("", encoding="utf-8") + assert diff_module.apply_diff_to_file(file_path, "", "a\n") is False + + class Boom(Exception): + pass + + def raise_patch(_text): + raise Boom("fail") + + monkeypatch.setattr(diff_module, "PatchSet", raise_patch) + assert diff_module.apply_diff_to_file(file_path, "a\n", "b\n") is False diff --git a/uv.lock b/uv.lock index 2ab118f..3a484aa 100644 --- a/uv.lock +++ b/uv.lock @@ -175,7 +175,6 @@ toml = [ [[package]] name = "dbml-to-sqlmodel" -version = "0.1.0" source = { editable = "." } dependencies = [ { name = "pydantic" }, @@ -191,6 +190,7 @@ runtime = [ { name = "aiosqlite" }, { name = "fastapi", extra = ["all"] }, { name = "fastcrud" }, + { name = "greenlet" }, { name = "sqladmin" }, { name = "sqlmodel" }, ] @@ -209,6 +209,7 @@ requires-dist = [ { name = "aiosqlite", marker = "extra == 'runtime'", specifier = ">=0.22.1" }, { name = "fastapi", extras = ["all"], marker = "extra == 'runtime'", specifier = ">=0.128.0" }, { name = "fastcrud", marker = "extra == 'runtime'", specifier = ">=0.20.1" }, + { name = "greenlet", marker = "extra == 'runtime'", specifier = ">=3.0.0" }, { name = "pydantic", specifier = ">=2.11.1" }, { name = "pydbml", specifier = ">=1.2.1" }, { name = "questionary", specifier = ">=2.0.0" },