syntax

Author: vishalshenoy

examples/sqlalchemy_1.6_to_2.0/README.md

@@ -60,23 +60,10 @@ The migration script handles four key transformations:
        addresses: Mapped[List["Address"]] = relationship()
    ```
 
-## Running the Example
-
-```bash
-# Install Codegen
-pip install codegen
-
-# Run the migration
-python run.py
-```
-
-The script will process all Python files in the `repo-before` directory and apply the transformations in the correct order.
-
 ## Understanding the Code
 
-- `run.py` - The migration script
-- `repo-before/` - Sample SQLAlchemy 1.6 application to migrate
-- `guide.md` - Additional notes and explanations
+- `input_repo` - Sample SQLAlchemy 1.6 application to migrate
+- `output_repo` - Sample SQLAlchemy 2.0 application after migration
 
 ## Learn More
 

examples/sqlalchemy_1.6_to_2.0/output_repo/database.py

@@ -0,0 +1,10 @@
+# database.py
+from sqlalchemy import create_engine
+from sqlalchemy.orm import sessionmaker
+
+SQLALCHEMY_DATABASE_URL = "postgresql://user:password@localhost/dbname"  # Change to your database URL
+
+engine = create_engine(SQLALCHEMY_DATABASE_URL)
+SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
+
+Base = declarative_base()

examples/sqlalchemy_1.6_to_2.0/output_repo/main.py

@@ -0,0 +1,102 @@
+# main.py
+from fastapi import FastAPI, Depends, HTTPException
+from sqlalchemy.orm import Session
+import models
+import schemas
+from database import SessionLocal, engine
+from typing import List
+
+# Initialize the app and create database tables
+app = FastAPI()
+models.Base.metadata.create_all(bind=engine)
+
+# Dependency for the database session
+def get_db():
+    db = SessionLocal()
+    try:
+        yield db
+    finally:
+        db.close()
+
+# CRUD Operations
+
+@app.post("/books/", response_model=schemas.Book)
+def create_book(book: schemas.BookCreate, db: Session = Depends(get_db)):
+    db_book = models.Book(**book.dict())
+    db.add(db_book)
+    db.commit()
+    db.refresh(db_book)
+    return db_book
+
+@app.get("/books/", response_model=List[schemas.Book])
+def read_books(skip: int = 0, limit: int = 10, db: Session = Depends(get_db)):
+    books = db.query()(models.Book).offset(skip).limit(limit).scalars().all()
+    return books
+
+@app.get("/books/{book_id}", response_model=schemas.Book)
+def read_book(book_id: int, db: Session = Depends(get_db)):
+    book = db.query()(models.Book).where(models.Book.id == book_id).first()
+    if book is None:
+        raise HTTPException(status_code=404, detail="Book not found")
+    return book
+
+@app.put("/books/{book_id}", response_model=schemas.Book)
+def update_book(book_id: int, book: schemas.BookCreate, db: Session = Depends(get_db)):
+    db_book = db.query()(models.Book).where(models.Book.id == book_id).first()
+    if db_book is None:
+        raise HTTPException(status_code=404, detail="Book not found")
+    for key, value in book.dict().items():
+        setattr(db_book, key, value)
+    db.commit()
+    db.refresh(db_book)
+    return db_book
+
+@app.delete("/books/{book_id}", response_model=schemas.Book)
+def delete_book(book_id: int, db: Session = Depends(get_db)):
+    db_book = db.query()(models.Book).where(models.Book.id == book_id).first()
+    if db_book is None:
+        raise HTTPException(status_code=404, detail="Book not found")
+    db.delete(db_book)
+    db.commit()
+    return db_book
+
+@app.post("/publishers/", response_model=schemas.Publisher)
+def create_publisher(publisher: schemas.PublisherCreate, db: Session = Depends(get_db)):
+    db_publisher = models.Publisher(**publisher.dict())
+    db.add(db_publisher)
+    db.commit()
+    db.refresh(db_publisher)
+    return db_publisher
+
+@app.get("/publishers/", response_model=List[schemas.Publisher])
+def read_publishers(skip: int = 0, limit: int = 10, db: Session = Depends(get_db)):
+    publishers = db.query()(models.Publisher).offset(skip).limit(limit).scalars().all()
+    return publishers
+
+@app.get("/publishers/{publisher_id}", response_model=schemas.Publisher)
+def read_publisher(publisher_id: int, db: Session = Depends(get_db)):
+    publisher = db.query()(models.Publisher).where(models.Publisher.id == publisher_id).first()
+    if not publisher:
+        raise HTTPException(status_code=404, detail="Publisher not found")
+    return publisher
+
+@app.put("/publishers/{publisher_id}", response_model=schemas.Publisher)
+def update_publisher(publisher_id: int, publisher: schemas.PublisherCreate, db: Session = Depends(get_db)):
+    db_publisher = db.query()(models.Publisher).where(models.Publisher.id == publisher_id).first()
+    if not db_publisher:
+        raise HTTPException(status_code=404, detail="Publisher not found")
+    for key, value in publisher.dict().items():
+        setattr(db_publisher, key, value)
+    db.commit()
+    db.refresh(db_publisher)
+    return db_publisher
+
+@app.delete("/publishers/{publisher_id}", response_model=schemas.Publisher)
+def delete_publisher(publisher_id: int, db: Session = Depends(get_db)):
+    db_publisher = db.query()(models.Publisher).where(models.Publisher.id == publisher_id).first()
+    if not db_publisher:
+        raise HTTPException(status_code=404, detail="Publisher not found")
+    db.delete(db_publisher)
+    db.commit()
+    return db_publisher
+

examples/sqlalchemy_1.6_to_2.0/output_repo/models.py

@@ -0,0 +1,32 @@
+from typing import List, Optional
+from sqlalchemy import Integer, String, ForeignKey
+from sqlalchemy.orm import relationship, Mapped, mapped_column
+from database import Base
+
+class Publisher(Base):
+    __tablename__ = "publishers"
+
+    id: Mapped[int] = mapped_column(Integer, primary_key=True, index=True)
+    name: Mapped[str] = mapped_column(String, unique=True, index=True)
+    books: Mapped[List["Book"]] = relationship(
+        "Book", 
+        back_populates="publisher",
+        lazy='selectin'
+    )
+
+class Book(Base):
+    __tablename__ = "books"
+
+    id: Mapped[int] = mapped_column(Integer, primary_key=True, index=True)
+    title: Mapped[str] = mapped_column(String, index=True)
+    author: Mapped[str] = mapped_column(String, index=True)
+    description: Mapped[Optional[str]] = mapped_column(String, nullable=True)
+    publisher_id: Mapped[Optional[int]] = mapped_column(
+        Integer, 
+        ForeignKey("publishers.id"),
+        nullable=True
+    )
+    publisher: Mapped[Optional["Publisher"]] = relationship(
+        "Publisher", 
+        back_populates="books"
+    )

examples/sqlalchemy_1.6_to_2.0/output_repo/schemas.py

@@ -0,0 +1,31 @@
+from pydantic import BaseModel
+from typing import List, Optional
+
+class PublisherBase(BaseModel):
+    name: str
+
+class PublisherCreate(PublisherBase):
+    pass
+
+class Publisher(PublisherBase):
+    id: int
+    books: List["Book"] = []
+
+    class Config:
+        orm_mode = True
+
+class BookBase(BaseModel):
+    title: str
+    author: str
+    description: str
+    publisher_id: Optional[int]
+
+class BookCreate(BookBase):
+    pass
+
+class Book(BookBase):
+    id: int
+    publisher: Optional[Publisher]
+
+    class Config:
+        orm_mode = True