Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
async def test_iterate_outside_transaction_with_temp_table(database_url):
"""
Same as test_iterate_outside_transaction_with_values but uses a
temporary table instead of a list of values.
"""
database_url = DatabaseURL(database_url)
if database_url.dialect == "sqlite":
pytest.skip("SQLite interface does not work with temporary tables.")
async with Database(database_url) as database:
query = "CREATE TEMPORARY TABLE no_transac(num INTEGER)"
await database.execute(query)
query = "INSERT INTO no_transac(num) VALUES (1), (2), (3), (4), (5)"
await database.execute(query)
query = "SELECT * FROM no_transac"
iterate_results = []
async for result in database.iterate(query=query):
iterate_results.append(result)
assert len(iterate_results) == 5
import databases
import orm
import sqlalchemy
from orm.models import QuerySet
from starlette.authentication import BaseUser
from .resources import hasher
database = databases.Database("sqlite:///tests/test.db", force_rollback=True)
metadata = sqlalchemy.MetaData()
class UserQuerySet(QuerySet):
async def create_user(self, *, password: str, **kwargs):
kwargs["password"] = await hasher.make(password)
return await self.create(**kwargs)
class User(BaseUser, orm.Model):
__tablename__ = "user"
__database__ = database
__metadata__ = metadata
objects = UserQuerySet()
async def test_concurrent_access_on_single_connection(database_url):
database_url = DatabaseURL(database_url)
if database_url.dialect != "postgresql":
pytest.skip("Test requires `pg_sleep()`")
async with Database(database_url, force_rollback=True) as database:
async def db_lookup():
await database.fetch_one("SELECT pg_sleep(1)")
await asyncio.gather(db_lookup(), db_lookup())
from starlette.authentication import BaseUser, requires
from starlette.middleware.authentication import AuthenticationMiddleware
from starlette.requests import Request
from starlette.responses import JSONResponse, PlainTextResponse
from starlette_auth_toolkit.backends import MultiAuth
from starlette_auth_toolkit.base.backends import BaseTokenAuth
from starlette_auth_toolkit.contrib.orm import ModelBasicAuth
from starlette_auth_toolkit.cryptography import (
PBKDF2Hasher,
generate_random_string,
)
# Database setup
database = databases.Database(
os.getenv("DATABASE_URL"), force_rollback=os.getenv("TESTING")
)
metadata = sqlalchemy.MetaData()
class UserQuerySet(QuerySet):
async def create_user(self, *, password: str, **kwargs):
kwargs["password"] = await hasher.make(password)
return await self.create(**kwargs)
class User(BaseUser, orm.Model):
__tablename__ = "user"
__database__ = database
__metadata__ = metadata
async def sqlalchemy_user_db() -> AsyncGenerator[SQLAlchemyUserDatabase, None]:
Base: DeclarativeMeta = declarative_base()
class User(SQLAlchemyBaseUserTable, Base):
pass
DATABASE_URL = "sqlite:///./test.db"
database = Database(DATABASE_URL)
engine = sqlalchemy.create_engine(
DATABASE_URL, connect_args={"check_same_thread": False}
)
Base.metadata.create_all(engine)
await database.connect()
yield SQLAlchemyUserDatabase(database, User.__table__)
Base.metadata.drop_all(engine)
from starlette.staticfiles import StaticFiles
from starlette.templating import Jinja2Templates
from source import settings
import databases
import httpx
templates = Jinja2Templates(directory="templates")
statics = StaticFiles(directory="statics")
if settings.TESTING:
database = databases.Database(settings.TEST_DATABASE_URL, force_rollback=True)
else: # pragma: nocover
database = databases.Database(settings.DATABASE_URL)
def url_for(*args, **kwargs):
from source.app import app
return app.url_path_for(*args, **kwargs)
import typing
import databases
import orm
import sqlalchemy
from .parser import Parser
database = databases.Database("sqlite:///guitarlette.db")
metadata = sqlalchemy.MetaData()
class Song(orm.Model):
__tablename__ = "songs"
__database__ = database
__metadata__ = metadata
id = orm.Integer(primary_key=True)
title = orm.String(max_length=100)
artist = orm.String(max_length=100)
content = orm.Text(allow_null=True)
def get_dict(self, new: bool = False) -> typing.Dict:
html = Parser(self.content).html
data = {
import asyncio
from typing import Tuple, List
import databases
import orm
import sqlalchemy
from stilio.config import DATABASE_URL
database = databases.Database(DATABASE_URL)
metadata = sqlalchemy.MetaData()
class Torrent(orm.Model):
__tablename__ = "torrent"
__database__ = database
__metadata__ = metadata
id = orm.Integer(primary_key=True)
info_hash = orm.String(max_length=40, unique=True, index=True)
name = orm.String(max_length=512)
date = orm.DateTime()
def __str__(self):
return self.name
async def query(self, query):
async with Database(self.database_url) as database:
return await database.fetch_all(query=q)
import databases
import sqlalchemy
from fastapi import FastAPI
from fastapi_users import BaseUser, FastAPIUsers
from fastapi_users.authentication import JWTAuthentication
from fastapi_users.db import SQLAlchemyBaseUserTable, SQLAlchemyUserDatabase
from sqlalchemy.ext.declarative import DeclarativeMeta, declarative_base
DATABASE_URL = "sqlite:///./test.db"
SECRET = "SECRET"
database = databases.Database(DATABASE_URL)
Base: DeclarativeMeta = declarative_base()
class UserTable(Base, SQLAlchemyBaseUserTable):
pass
engine = sqlalchemy.create_engine(
DATABASE_URL, connect_args={"check_same_thread": False}
)
Base.metadata.create_all(engine)
users = UserTable.__table__
user_db = SQLAlchemyUserDatabase(database, users)