Added examples

Author: Ewen-Zippedscript

examples/legacy/app.py

@@ -0,0 +1,83 @@
+from typing import Optional
+
+from fastapi import FastAPI
+from models import User, db
+from pydantic import BaseModel
+
+from fastapi_sqlalchemy import DBSessionMiddleware
+
+app = FastAPI()
+
+# Add DB session middleware with db_url specified
+app.add_middleware(DBSessionMiddleware, db_url="sqlite:///example.db")
+
+
+# Endpoint to retrieve all users
+@app.get("/users")
+def get_users():
+    """
+    Retrieve a list of all users.
+
+    Returns:
+        List[User]: A list of User objects.
+    """
+    return db.session.query(User).all()
+
+
+# Pydantic model for creating new users
+class UserCreate(BaseModel):
+    name: str
+    email: str
+
+
+# Endpoint to add a new user
+@app.post("/add_user")
+def add_user(user_data: UserCreate):
+    """
+    Add a new user to the database.
+
+    Args:
+        user_data (UserCreate): User data including name and email.
+
+    Returns:
+        dict: A message indicating the success of the operation.
+    """
+    user = User(**user_data.dict())
+    db.session.add(user)
+    db.session.commit()
+    return {"message": "User created successfully"}
+
+
+# Pydantic model for updating user information
+class UserUpdate(UserCreate):
+    id: int
+    name: Optional[str]
+    email: Optional[str]
+
+
+# Endpoint to update user information
+@app.post("/update_user")
+def update_user(user_data: UserUpdate):
+    """
+    Update user information in the database.
+
+    Args:
+        user_data (UserUpdate): User data including ID, name, and email for updating.
+
+    Returns:
+        dict: A message indicating the success of the operation.
+    """
+    user = db.session.query(User).filter_by(id=user_data.id).first()
+    if user_data.name:
+        user.name = user_data.name
+    if user_data.email:
+        user.email = user_data.email
+    db.session.add(user)
+    db.session.commit()
+    return {"message": "User updated successfully"}
+
+
+if __name__ == "__main__":
+    import uvicorn
+
+    uvicorn.run(app, host="127.0.0.1", port=8000)

examples/legacy/models.py

@@ -0,0 +1,17 @@
+from sqlalchemy import Column, Integer, String, create_engine
+from sqlalchemy.orm import declarative_base, sessionmaker
+
+from fastapi_sqlalchemy import db
+
+
+# Define the User class representing the "users" database table
+# Using the SQLAlchemy Base property instead of defining your own
+class User(db.Base):
+    __tablename__ = "users"
+
+    id = Column(Integer, primary_key=True)
+    name = Column(String)
+    email = Column(String)
+
+    def __repr__(self):
+        return f"User(id={self.id}, name='{self.name}',email='{self.email}')"

examples/multi_db/app.py

@@ -0,0 +1,118 @@
+# Import necessary modules and classes
+from typing import Optional
+
+from fastapi import FastAPI
+from models.posts import Post, post_db  # Import the Post model and post_db database instance
+from models.users import User, user_db  # Import the User model and user_db database instance
+from pydantic import BaseModel
+
+from fastapi_sqlalchemy import (
+    DBSessionMiddleware,  # Import the DBSessionMiddleware for database sessions
+)
+
+# Create a FastAPI application instance
+app = FastAPI()
+
+# Add the DBSessionMiddleware as a middleware to the FastAPI app, connecting it to the specified databases
+app.add_middleware(DBSessionMiddleware, db=[post_db, user_db])
+
+
+# Define an endpoint for retrieving all users
+@app.get("/users")
+def get_users():
+    """
+    Endpoint to retrieve a list of all users.
+    Returns:
+        List[User]: A list of User objects representing all users in the database.
+    """
+    return User.get_all()
+
+
+# Define a Pydantic model for creating a new user
+class UserCreate(BaseModel):
+    name: str
+    email: str
+
+
+# Define an endpoint for adding a new user
+@app.post("/add_user")
+def add_user(user_data: UserCreate):
+    """
+    Endpoint to add a new user to the database.
+    Args:
+        user_data (UserCreate): User data provided in the request body.
+    Returns:
+        dict: A message indicating the success of the operation.
+    """
+    user = User(**user_data.dict())
+    user.save()
+    return {"message": "User created successfully"}
+
+
+# Define a Pydantic model for updating user information
+class UserUpdate(UserCreate):
+    id: int
+    name: Optional[str]
+    email: Optional[str]
+
+
+# Define an endpoint for updating user information
+@app.post("/update_user")
+def update_user(user_data: UserUpdate):
+    """
+    Endpoint to update user information in the database.
+    Args:
+        user_data (UserUpdate): User data provided in the request body.
+    Returns:
+        dict: A message indicating the success of the operation.
+    """
+    user = User.get(id=user_data.id)
+    user.update(**user_data.dict())
+    user.save()
+    return {"message": "User updated successfully"}
+
+
+# Define a Pydantic model for retrieving posts by user ID
+class UserPosts(BaseModel):
+    user_id: int
+
+
+# Define an endpoint for retrieving posts by user ID
+@app.get("/posts")
+def get_posts(user: UserPosts):
+    """
+    Endpoint to retrieve posts by a specific user ID.
+    Args:
+        user (UserPosts): User ID provided in the query parameters.
+    Returns:
+        List[Post]: A list of Post objects belonging to the specified user.
+    """
+    posts = Post.get_all(user_id=user.user_id)
+    return posts
+
+
+# Define a Pydantic model for creating a new post
+class PostCreate(UserPosts):
+    title: str
+    content: str
+
+
+# Define an endpoint for adding a new post
+@app.post("/add_post")
+def add_post(post_data: PostCreate):
+    """
+    Endpoint to add a new post to the database.
+    Args:
+        post_data (PostCreate): Post data provided in the request body.
+    Returns:
+        dict: A message indicating the success of the operation.
+    """
+    post = Post(**post_data.dict())
+    post.save()
+    return {"message": "Post created successfully"}
+
+
+if __name__ == "__main__":
+    import uvicorn
+
+    uvicorn.run(app, host="127.0.0.1", port=8000)

examples/multi_db/models/__init__.py

@@ -0,0 +1,52 @@
+import inspect
+from typing import Dict, List
+
+from sqlalchemy import Column
+
+from fastapi_sqlalchemy import ModelBase
+
+
+class BaseModel(ModelBase):
+    @classmethod
+    def new(cls, **kwargs):
+        obj = cls(**kwargs)
+        obj.save()
+        return obj
+
+    @classmethod
+    def get(cls, **kwargs):
+        result: cls = cls.query.filter_by(**kwargs).first()
+        return result
+
+    @classmethod
+    def get_all(cls, **kwargs):
+        result: List[cls] = cls.query.filter_by(**kwargs).all()
+        return result
+
+    def update(self, **kwargs):
+        for column, value in kwargs.items():
+            setattr(self, column, value)
+
+        self.save()
+        return self
+
+    def __repr__(self):
+        try:
+            columns = dict(
+                (column.name, getattr(self, column.name)) for column in self.__table__.columns
+            )
+
+        except:
+            o = {}
+            members = inspect.getmembers(self)
+            for name, obj in members:
+                if type(obj) == Column:
+                    o[name] = obj
+            columns = o
+
+        column_strings = []
+        for column, value in columns.items():
+            column_strings.append(f"{column}: {value}")
+
+        repr = f"<{self.__class__.__name__} {', '.join(column_strings)}>"
+        return repr

examples/multi_db/models/posts.py

@@ -0,0 +1,20 @@
+from sqlalchemy import Column, Integer, MetaData, String, create_engine
+from sqlalchemy.orm import DeclarativeMeta, declarative_base, sessionmaker
+
+from fastapi_sqlalchemy import SQLAlchemy
+
+from . import BaseModel
+
+# Create a SQLAlchemy instance with a connection to the SQLite database "post.db"
+post_db = SQLAlchemy("sqlite:///post.db")
+
+
+# Define the User class representing the "posts" database table
+# using the SQLAlchemy Base property instead of defining your own
+# And inheriting from the BaseModel class for type hinting and helpful builtin methods and properties
+class Post(BaseModel, post_db.Base):
+    __tablename__ = "posts"
+    id = Column(Integer, primary_key=True)
+    title = Column(String)
+    content = Column(String)
+    user_id = Column(Integer)

examples/multi_db/models/users.py

@@ -0,0 +1,36 @@
+# Import necessary modules and classes
+from sqlalchemy import Column, Integer, String, create_engine  # Import SQLAlchemy components
+from sqlalchemy.orm import (  # Import SQLAlchemy components
+    DeclarativeMeta,
+    declarative_base,
+    sessionmaker,
+)
+
+from fastapi_sqlalchemy import SQLAlchemy  # Import the SQLAlchemy extension
+
+from . import BaseModel  # Import the custom BaseModel
+
+# Create a SQLAlchemy instance with a connection to the SQLite database "user.db"
+user_db = SQLAlchemy("sqlite:///user.db")
+
+
+# Define the User class representing the "users" database table
+# Using the SQLAlchemy Base property instead of defining your own
+# And inheriting from the BaseModel class for type hinting and helpful builtin methods and properties
+class User(BaseModel, user_db.Base):
+    """
+    Represents a user in the database.
+
+    Attributes:
+        id (int): The primary key of the user.
+        name (str): The name of the user.
+        email (str): The email address of the user.
+    """
+
+    # Name of the database table associated with this class
+    __tablename__ = "users"
+
+    # Columns corresponding to the attributes of the User class
+    id = Column(Integer, primary_key=True)
+    name = Column(String)
+    email = Column(String)