Skip to content

Instantly share code, notes, and snippets.

@emattiza

emattiza/.envrc

Last active November 23, 2022 14:46
Show Gist options

  • Select an option

    Learn more about clone URLs
  • Save emattiza/af4a571848c820d04e9f941b49795d60 to your computer and use it in GitHub Desktop.

Select an option

Learn more about clone URLs
Save emattiza/af4a571848c820d04e9f941b49795d60 to your computer and use it in GitHub Desktop.
Download ZIP
Raw
.envrc
use flake
Raw
.gitignore
.direnv
__pycache__
Raw
app.py
#!/usr/bin/env nix-shell
#!nix-shell -I nixpkgs=https://github.com/nixos/nixpkgs/archive/nixos-unstable.tar.gz
#!nix-shell shell.nix
#!nix-shell --run "uvicorn app:app --host 0.0.0.0 --reload"
from asyncio import current_task
from fastapi.responses import JSONResponse
from sqlalchemy import Column, Integer, String, Boolean, delete, update
from sqlalchemy.future import select
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.ext.asyncio import (
AsyncSession,
async_scoped_session,
create_async_engine,
)
from sqlalchemy.orm import sessionmaker
from fastapi import Depends, FastAPI, status
from pydantic import BaseModel
from typing import Literal, Optional, List
# -----------------------------------------------------------------------
# Database init stuff
# -----------------------------------------------------------------------
SQLALCHEMY_DATABASE_URL = "sqlite+aiosqlite:///test.db"
engine = create_async_engine(
SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False}
)
# engine = create_engine(
# SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False}
# )
SessionLocal = sessionmaker(
bind=engine,
expire_on_commit=False,
class_=AsyncSession,
autocommit=False,
autoflush=True,
)
Base = declarative_base()
# -----------------------------------------------------------------------
# Database Models
# -----------------------------------------------------------------------
class Entry(Base):
__tablename__ = "entries"
id = Column(Integer, primary_key=True)
title = Column(String)
order = Column(Integer)
completed = Column(Boolean)
def __init__(self, title=None, order=None, completed=False):
self.title = title
self.order = order
self.completed = completed
def __repr__(self):
return "<Entry: {}>".format(self.title)
# -----------------------------------------------------------------------
# Pydantic Models
# -----------------------------------------------------------------------
class HelloWorld(BaseModel):
message: Literal["Hello, world!"]
class TodoEntryBase(BaseModel):
title: str
order: int
class TodoCreate(BaseModel):
title: str
order: int
completed: bool
class TodoEntry(TodoEntryBase):
id: Optional[int]
completed: Optional[bool]
class Config:
orm_mode = True
# -----------------------------------------------------------------------
# Todo Managers
# -----------------------------------------------------------------------
class TodoDAL:
def __init__(self, db_session: AsyncSession):
self.db_session = db_session
async def get_todos(self, skip: int = 0, limit: int = 100) -> List[TodoEntry]:
todos = await self.db_session.execute(select(Entry).offset(skip).limit(limit))
return todos.scalars().all()
async def create_todo(self, todo: TodoCreate) -> TodoEntry:
new_todo = Entry(**todo.dict())
self.db_session.add(new_todo)
await self.db_session.flush()
return TodoEntry(**new_todo.__dict__).json()
async def update_todo(self, todo: TodoEntry) -> TodoEntry:
stmt = update(Entry).where(Entry.id == todo.id).returning(Entry)
orm_stmt = (
select(Entry).from_statement(stmt).execution_options(populate_existing=True)
)
todo = await self.db_session.execute(orm_stmt)
return todo
async def delete_todo(self, todo: TodoEntry):
stmt = delete(Entry).where(Entry.id == todo.id)
todo = await self.db_session.execute(stmt)
async def get_todo_dal():
async with async_scoped_session(SessionLocal, scopefunc=current_task)() as session:
async with session.begin():
yield TodoDAL(session)
# -----------------------------------------------------------------------
# App init
# -----------------------------------------------------------------------
app = FastAPI()
@app.on_event("startup")
async def startup():
# create db tables
async with engine.begin() as conn:
await conn.run_sync(Base.metadata.drop_all)
await conn.run_sync(Base.metadata.create_all)
# -----------------------------------------------------------------------
# App views
# -----------------------------------------------------------------------
@app.get("/")
async def index():
return "Hello!"
@app.get("/todos", response_model=list[TodoEntry])
async def todos_list(
skip: int = 0, limit: int = 100, todo_dal: TodoDAL = Depends(get_todo_dal)
):
todos = await todo_dal.get_todos(skip, limit)
return todos
@app.post("/todos", response_model=TodoEntry)
async def create_todo(todo: TodoCreate, todo_dal: TodoDAL = Depends(get_todo_dal)):
item = await todo_dal.create_todo(todo)
return JSONResponse(status_code=status.HTTP_201_CREATED, content=item)
@app.patch("/todos")
async def update_todo(todo: TodoEntry, todo_dal: TodoDAL = Depends(get_todo_dal)):
updated = await todo_dal.update_todo(todo)
return updated
@app.delete("/todos")
async def delete_todo(todo: TodoEntry, todo_dal: TodoDAL = Depends(get_todo_dal)):
await todo_dal.delete_todo(todo)
Raw
flake.lock
{
"nodes": {
"flake-utils": {
"locked": {
"lastModified": 1659877975,
"narHash": "sha256-zllb8aq3YO3h8B/U0/J1WBgAL8EX5yWf5pMj3G0NAmc=",
"owner": "numtide",
"repo": "flake-utils",
"rev": "c0e246b9b83f637f4681389ecabcb2681b4f3af0",
"type": "github"
},
"original": {
"owner": "numtide",
"repo": "flake-utils",
"type": "github"
}
},
"nixpkgs": {
"locked": {
"lastModified": 1665940183,
"narHash": "sha256-cPe3F7CtnxU9YbJpc3Adl4d9kX+turqTv5FxM98i8vg=",
"owner": "NixOS",
"repo": "nixpkgs",
"rev": "104e8082de1b20f9d0e1f05b1028795ed0e0e4bc",
"type": "github"
},
"original": {
"owner": "NixOS",
"ref": "nixos-unstable",
"repo": "nixpkgs",
"type": "github"
}
},
"root": {
"inputs": {
"flake-utils": "flake-utils",
"nixpkgs": "nixpkgs"
}
}
},
"root": "root",
"version": 7
}
Raw
flake.nix
{
description = "my portable dotfiles";
inputs = {
flake-utils.url = "github:numtide/flake-utils";
nixpkgs.url = "github:NixOS/nixpkgs/nixos-unstable";
};
outputs = {
self,
nixpkgs,
flake-utils,
}:
flake-utils.lib.eachDefaultSystem (system: let
pkgs = nixpkgs.legacyPackages.${system};
in rec {
devShells = {
default = import ./shell.nix {inherit pkgs;};
};
devShell = devShells.default;
});
}
Raw
shell.nix
{pkgs ? import <nixpkgs> {}, ...}: let
pythonPkgs = p:
with p; [
fastapi
uvicorn
orjson
sqlalchemy
aiosqlite
factory_boy
];
python-env = pkgs.python310.withPackages pythonPkgs;
in
pkgs.mkShell {
buildInputs = with pkgs; [
python-env
pyright
black
starship
];
shellHook = ''
#source <(${pkgs.starship}/bin/starship init bash --print-full-init)
'';
}
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment