FastAPI使用Depends()注入返回或产生的变量。例如，
# Dependency
def get_db():
db = SessionLocal()
try:
yield db
finally:
db.close()
...
def create_user(db: Session = Depends(get_db)):
...
如果我想在其他地方(在FastAPI路由之外)使用该get_db()，我该如何做？我知道这是Python的核心知识，但我似乎搞不懂。我最初的想法是db = yield from get_db()，但我不能
浏览 4提问于2021-03-10得票数 2
1回答
如果我想在处理请求时使用数据库，我会像这样做一个依赖注入： @app.post("/sample_test")
async def sample_test(db: Session = Depends(get_db)):
return db.query(models.User.height).all() 但我不能在这样的事件中这样做： @app.on_event("startup")
async def sample_test(db: Session = Depends(get_db)):
return db.query(models.User.
浏览 24提问于2021-08-05得票数 2
回答已采纳
1回答
给出了定义如下的脓性BaseModel类：
from typing import List, Optional
from uuid import uuid4
from pydantic import BaseModel, Field
from server.database import get_db
class Campaign(BaseModel):
id: Optional[str] = Field(default_factory=lambda: str(uuid4()))
name: str
@staticmethod
async def all
浏览 1提问于2021-10-11得票数 0
回答已采纳
现在，我尝试在mongoldb中使用pymongo插入数据。
get_db().users.update({'_id':ObjectId(session['user_id'])},{'$push':{'hme':ObjectId(id)}},upsert=True)
但是，该方法会产生重复的ObjectID.before尝试find_one()。
if not ObjectId(id) in get_db().users.find_one({'_id':ObjectId(session['user_id&#
浏览 0提问于2015-06-25得票数 0
因此，正如标题所描述的，python没有提交到数据库，即使我要求它
下面的代码是我用来插入数据库的代码：
try:
get_db().execute("insert into Profesor(DNI, NombreApellidos, Telefono, Email, Ranking) values(?,?,?,?,?)", [dni, nombre, telefono, email, ranking])
get_db().commit()
get_db().close()
return "OK"
except:
re
浏览 3提问于2020-12-10得票数 0
在快速API中进行单元测试时，我很难访问SQLAlchemy。
def test_successful_register_ios():
response = client.post("/register/",headers={},json={})
assert response.status_code == 201
device = get_db().query(Device).get("a1")
assert device.expires_at == mydate
AttributeError：“生成器”对象没有属性“查询”
浏览 4提问于2021-01-31得票数 11
1回答
为什么这段代码仍然将用户添加到sqlite3 db，即使len(cursor.fetchall())大于0？
#just for refrence, the issue is in the next block of code
def get_db():
db = getattr(g, '_database', None)
if db is None:
db = g._database = sqlite3.connect(DATABASE)
return db
@app.route("/add_user", method
浏览 2提问于2019-06-30得票数 0
回答已采纳
1回答
我有一个长时间运行的查询，我尝试通过一个未缓冲的pdo-语句(参见下面的代码)来获取这个查询。但是，当我“执行”查询时没有缓冲，或者我做了一个'fetchAll'，时间(也见下面)并没有真正改变太多.我的想法是，非缓冲只执行查询并给我一个游标？
get_db函数返回一个Zend_Db对象。
l('Start 1');
$sql = get_db('ATO')->select()
... Big query...
->assemble(
浏览 3提问于2012-07-03得票数 0
我希望rubymine能够自动补全，比如方法。
def get_db
return Mysql::new(DB_HOST, "x", "x", "x")
end
DB = get_db
如果我在哪里输入
DB.
我应该得到Mysql对象的所有方法。我需要做些什么才能让RubyMine正确地做到这一点吗？
浏览 1提问于2013-04-24得票数 1
4回答
编辑:我在问全局变量在像tornado这样的单线程web框架中是否安全
我正在使用mongoengine orm，它从全局变量获取数据库连接：
_get_db() # gets the db connection
我也在使用tornado，一个单线程的python web框架。在一个特定的视图中，我需要获取一个数据库连接并取消引用一个类似于外键的DBRef对象
# dereference a DBRef
_get_db().dereference(some_db_ref)
由于_get_db返回的连接是一个全局变量，那么是否有可能发生冲突，并将错误的值返回给错误的线程？
浏览 0提问于2010-02-12得票数 5
回答已采纳
1回答
我正在尝试使用以下代码将图片加载到mysql数据库中：
cursor, db = get_db()
cursor.execute("UPDATE People SET photo = LOAD_FILE(\'myphoto.jpg\')")
cursor.close()
db.commit()
db.close()
我已经能够将照片设置为其他值，方法是将LOAD_FILE短语替换为更简单的值，因此我知道问题源于加载图片。同时，myphoto.jpg位于运行代码的目录中。MySql执行这些命令时不会出错，但实际上不会将图片放入数据库中。可能会出什么问题，我该如何修
浏览 1提问于2016-08-29得票数 1
1回答
因此，为了学习目的，我开始为一个微框架编写一个引导脚本(两者都了解更多关于编程、php和oop的知识)，我遇到了这种奇怪的意外行为。
变量$config启动Config()类的一个新对象，在Bootstrap的__construct (它是公共的)中调用它，然后在Bootstrap的__destruct (也是公共的)中使用它。如下面所示，$config的变量本身是公共的，并在__construct之前声明。
现在，奇怪的是，我在$config中使用__destruct得到了一个通知和致命错误，它说变量不存在，而致命错误是在非对象上调用成员函数(因为$config不存在)。
这是脚本，希望有人
浏览 4提问于2015-12-31得票数 2
回答已采纳
3回答
代码点火器错误？
、、
谁能帮我，我是新来的CI，我不知道我做错了什么。
我有这种控制器
main
<?php if ( ! defined('BASEPATH')) exit('No direct script access allowed');
class main extends CI_Controller {
public function index() {
$this->home();
}
public function home() {
$this->load->helper('url');
$t
浏览 0提问于2013-06-12得票数 0
回答已采纳
在我的Flask应用程序中，我使用：from pymongo import MongoClient导入了pymongo
然后我调用connect_db-method：
db = get_db(connect_db())
方法编码如下：
def connect_db():
client = MongoClient("localhost", 27017)
return client
def get_db(client):
return client(DATABASE)
但是我得到了错误TypeError: 'MongoClient' obj
浏览 5提问于2015-05-22得票数 1
2回答
我使用以下代码将JSON数据插入到SQLite数据库中：
# POST request here
if request.headers['Content-Type'] == 'application/json':
db = get_db()
db.execute('insert into places (lat, long, address, name) values (?, ?, ?, ?)', [request.data[0], request.data[1], request.data[2], request.data[3]
浏览 2提问于2012-11-10得票数 0
回答已采纳
1回答
我正在尝试将数据插入到mysql服务器中，而当我尝试将数据添加到SQLalchemy会话中时，我收到错误消息"Generator object has no attribute add“。
db=get_db()
temp = schema.User(**filtered_dict)
insert_data=models.User(**temp.dict())
db = get_db()
db.add(insert_data)
db.commit()
db.refresh()
会话生成器：
def get_db():
db_session = sessionlocal()
浏览 0提问于2021-05-14得票数 1
1回答
这些天，当我使用python orm时，遇到了一个事务问题。我使用这个orm保存了两个图书实例，在这两个节省之间，我引发了一个例外，所以我只是没有将它们保存到数据库中，但是它不起作用。有人能解释原因吗？我对蟒蛇很陌生，谢谢。
此代码如下：
from peewee import *
def get_db():
return SqliteDatabase("test.db")
class Book(Model):
id = PrimaryKeyField()
name = CharField()
class Meta:
datab
浏览 2提问于2016-11-06得票数 1
回答已采纳
def get_db():
if 'db' not in g:
g.db = sqlite3.connect(
current_app.config['DATABASE'],
detect_types=sqlite3.PARSE_DECLTYPES
)
g.db.row_factory = sqlite3.Row
return g.db
你好，试着学习烧瓶。目前正在浏览它的文档，刚刚完成了一个简单应用程序的教程。出于某种原因，
浏览 3提问于2021-07-19得票数 0
回答已采纳
我正在尝试创建一个FastAPI和异步sqlalchemy。 get_db依赖导致了一个奇怪的TypeError: <async_generator object get_db at 0x7ff6d9d9aa60> is not a callable object问题。 下面是我的代码： db.py from typing import Generator
from .db.session import SessionLocal
async def get_db() -> Generator:
try:
db = SessionLocal()
浏览 63提问于2021-04-18得票数 2
回答已采纳
1回答
我使用以下get_db()函数来获取我的SQLite数据库：
def get_db():
top = _app_ctx_stack.top
if not hasattr(top, 'sqlite_db'):
top.sqlite_db = sqlite3.connect(app.config['DATABASE'])
return top.sqlite_db
我相信我在这里没有使用Python SQLite游标。如果我执行以下操作：
db = get_db()
db.execute('DELETE FROM tab
浏览 1提问于2012-11-15得票数 0
回答已采纳
我正在学习烧瓶框架，并遵循烧瓶。我已经遵循了本教程的每一步逐行。最后，我得到了错误"sqlite3.OperationalError OperationalError: no table: entries"。我在linux机器上，以前从未使用过sqlite。我不知道如何解决这个问题。下面是flaskr.py的代码
# all the imports
import os
import sqlite3
from flask import Flask, request, session, g, redirect, url_for, abort, render_template, fl
浏览 2提问于2014-05-31得票数 4
回答已采纳
1回答
我有一个Flask应用程序，它看起来像这样：
from flask import Flask
app = Flask(__name__)
from pymongo import MongoClient
def get_db():
c = MongoClient()
return c.mydb
@app.route("/")
def hello():
db = get_db()
db.foo.find_one(...)
...
if __name__ == "__main__":
app.run()
看起来它工
浏览 2提问于2013-01-28得票数 2
我想实现我自己的依赖注入，比如、Fastapi、、、()实际上不需要使用外部包或框架。将采取什么办法？代码示例将对我有帮助。提前谢谢。
from typing import Callable, Optional, Any
class Depends:
def __init__(self, dependencies= Optional[Callable[..., Any]]):
self.dependencies = dependencies
def get_db():
pass
def get_token():
浏览 13提问于2022-10-22得票数 0
回答已采纳
在FastAPI docs中，建议使用如下生成器函数设置SQLAlchemy数据库会话依赖关系： async def get_db():
db = DBSession()
try:
yield db
finally:
db.close() 我的问题是为什么finally块会被执行？我的印象是生成器函数在每次输出时都会暂停执行。一旦session对象超出作用域，是否应该在db.close()从未运行的情况下丢弃get_db的执行？
浏览 35提问于2021-11-01得票数 0
回答已采纳
1回答
我在一个web后端工作，它从数据库文件中读取，处理数据并返回一个json对象。我对flask以及flask应用程序中变量的使用方式并不了解。
正如您在下面看到的，我正在从wsgi文件调用flaskApp。我根据flask文档创建了"get_db()“函数，但使用此函数没有任何改进。
有没有一种方法可以只连接到数据库一次，而不是每次调用URL？
#file flaskApp.py
#!/usr/bin/env python3.6
from flask import Flask
...
def get_db():
if 'db' not in g:
浏览 15提问于2019-11-15得票数 0
回答已采纳
我正试图在我的FastAPI应用程序中测试我的数据库。
from fastapi.testclient import TestClient
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from app.main import app
from app.db import Base, get_db, get_session
SQLALCHEMY_DATABASE_URL = "sqlite:///./test.db"
engine = create_engine(
浏览 4提问于2022-04-26得票数 0
我目前正在尝试为以下情况编写错误处理程序。当数据库为--read-only并试图编写时，我希望能够捕获一个错误。
拒绝用户“readonly”@的WordPress数据库错误插入命令
引发此错误是因为数据库是一个--read-only，并且试图写入数据库。但是，这个错误应该被我的catch条件捕获。
function drools_request($data, $uid) {
try {
$db = _get_db();
$db->insert("requests", [
"uid" => $uid,
浏览 2提问于2019-09-19得票数 3
1回答
FastAPI依赖覆盖失败
、、、、
我正在使用MongoMock测试我的FastAPI应用程序，但是我可以“覆盖FastAPI的依赖性”。
create_test_data将使用MongoMock的客户端插入一些假测试数据，并返回task_id请求所需的测试数据。
from mongomock import MongoClient
from fastapi.testclient import TestClient
from pytest import fixture
from app import app
from core.utils.db_utils import get_db
from models.tasks impor
浏览 11提问于2022-04-18得票数 0
我有一个包含以下详细信息的MongoDB数据库：
数据库名：myFirstDB还存储在环境变量DB_NAME中
集合名称：myFirstCollection还存储在环境变量DB_COLLEC中
我有以下python代码：
db = get_db()，其中get_db()返回pyMongo客户端连接。
我想要创建这样一个集合对象：
users = db.(os.environ.get('DB_NAME')).(os.environ.get('DB_COLLEC'))
但这总是返回语法错误。
在引用运算符中使用env变量的替代方法是什么？
编辑:更新以获取更多详细信
浏览 2提问于2021-05-08得票数 0
我正在设置路障，试图将错误处理添加到PHP中。目前，我有一个try、try和catch来处理当$db试图将insert->写入数据库(即--read-only )时出现的错误。我得到以下错误：
PHP致命错误:未捕获错误:在wp-content/mu-plugins/stat-collector/StatCollectorFunctions.php中找不到‘StatCollector\ ’类
<?php
namespace StatCollector;
function drools_request($data, $uid) {
try {
$db = _get_
浏览 4提问于2019-09-20得票数 1
回答已采纳
我使用的是gunicorn运行的Flask应用程序，如果我直接在我的服务器上运行它，它工作得很好，但如果我在Docker中运行它，它就会失败，并出现以下错误
Exception in thread Thread-1:
Traceback (most recent call last):
File "/usr/lib64/python3.8/threading.py", line 932, in _bootstrap_inner
self.run()
File "/usr/lib64/python3.8/threading.py", line 87
浏览 0提问于2021-06-07得票数 0
2回答
我正试图通过容器注册将一个快速的API todo应用部署到Heroku上。当我建立码头形象并在我的本地运行它。我能够访问我在中的狂妄自大。但是当我部署到heroku时，我无法访问这个错误：错误: Exec格式错误
这是我的main.py
from fastapi import FastAPI,Depends
from fastapi.middleware.cors import CORSMiddleware
from db.db import get_db
from bson.objectid import ObjectId
from models.todo import Todo
app
浏览 6提问于2022-05-14得票数 1
1回答
我尝试在我的python脚本中使用psycog2。 我在db.py文件中有一个get_db()函数(我需要在sqlite和psql之间进行选择)： def get_db():
if 'db' not in g:
info_db = getDatabaseType()
if info_db['type'] != 'pgsql': # If wrong type is setup, use sqlite by default
g.db = sqlite3.connect(
浏览 19提问于2020-07-27得票数 1
回答已采纳
我有一个应用程序使用烧瓶-Restful，我不知道如何从应用程序上下文中获取数据库连接信息。
以下是我到目前为止所拥有的：
app.py:
....
from flask.ext.restful import reqparse, abort, Api, Resource
app = Flask(__name__)
app.config['DATABASE'] = 'my.db'
api = Api(app)
api.add_resource(Foo, '/')
foo.py
...
from flask.ext.restful import R
浏览 1提问于2014-08-19得票数 1
回答已采纳
我只是想让FastAPI单元测试和SQLAlchemy一起工作，但是我在测试数据库中创建的对象时遇到了问题。根据docs，我有以下设置。 main.py routes = [
APIRoute('/games/', views.games_list, name='index', response_class=HTMLResponse),
]
settings = Settings()
app = FastAPI(debug=settings.debug, routes=routes) views.py # Dependency
def get_db
浏览 22提问于2020-12-07得票数 0
2回答
我在嘲笑mongodb的时候需要帮助。我使用来模拟mongodb。
我的项目结构是：
-- my_mongo.py
-- code.py
-- my_test.py
my_mongo.py有：
from pymongo import MongoClient
def get_db():
client = MongoClient(os.environ['MONGODB_URI'])
db = client['my_db']
return db
def insert(id, data):
return get_db().re
浏览 2提问于2018-12-17得票数 3
我在Idiorm文档中找到这样的东西时遇到了问题。我试图做一个区分大小写的搜索，比如：
$article=ORM::for_table('articles')->where_like('content','%fOo%')->find_many();
但是我希望搜索是区分大小写的，我需要配置什么？
LE:下面是表的定义方式：
ORM::get_db()->exec( 'DROP TABLE IF EXISTS pw_article' );
ORM::get_db()->exec(
'
浏览 4提问于2014-03-30得票数 0
回答已采纳
2回答
是否有在Google中设置全局连接的建议？我目前使用的方法似乎有点粗糙，我正在这样做：
import pymysql
from flask import Flask, jsonify
# lazy loading?
db_conn = None
db_cursor = None
def get_db():
global db_conn, db_cursor
if (db_conn, db_cursor) == (None, None):
global_conn = pymysql.connect(user="x", password="
浏览 7提问于2022-07-24得票数 0
回答已采纳
我是整个Flask和web编程方面的新手，所以我试图理解框架的上下文和db操作的原则。现在，我想了解如何在不使用sql和其他ORM的情况下管理DB连接，只使用普通的sql。
因此，使用sqlite3 db的简单示例。我的测试项目的结构：
.
├── application
│ ├── __init__.py
│ ├── routes.py
│ └── templates
│ ├── base.html
│ ├── index.html
├── config.py
├── requirements.txt
├── run.py
└── test.db
init
浏览 0提问于2019-04-18得票数 2
回答已采纳
当我尝试“搜索功能”时，我得到了422个错误。这是我的密码：
路由器
@router.get('/search')
def search(query: Optional[str], db: Session = Depends(get_db)):
return librarian.search_book(query, db)
图书馆员
def search_book(query: Optional[str], db: Session = Depends(get_db)):
res = db.query(models.Book).filter(models.Book
浏览 16提问于2022-06-12得票数 1
1回答
我有两个py文件。database.py和myapp.py.在myapp.py中，我想调用database.py来创建连接，并执行mysql查询，但我不希望关闭到数据库的连接，因为我需要重用它。这是我的密码。
有人能给我一些建议吗？
database.py
import mysql.connector
from flask import g
def connect_db():
return mysql.connector.connect(user='root', password='password', database='test')
浏览 2提问于2013-09-10得票数 0
我目前正在努力为以下情况添加错误处理。当拥有--read-only访问权限时，当$db数据库尝试向其写入时，将引发错误。这会在WP引擎上导致以下错误。 WordPress database error INSERT command denied to user 'readonly'@'xx.xxx.xx.xx' for table 'responses' for query INSERT INTO `responses` 我不希望这个错误不破坏应用程序，所以我尝试添加错误处理。但是，当我抛出一个异常时，它不会被捕获。我知道WP_ERROR是不同
浏览 26提问于2019-09-21得票数 1
回答已采纳
1回答
我需要一些经验的帮助。我不知道该怎么做，完全模糊了。
我正在使用joomla，并且我有带有onclick="get_db()“函数的li选项卡。
<li><a href="#addpanel3" data-toggle="tab" onclick="get_db()"><s1><?php echo Jtext::_('ST_OTHER_INFORMATION');?></s1></a></li>
下面是get_db()的函数
functio
浏览 0提问于2013-11-13得票数 0
1回答
由于某些原因，我无法传递一个参数，并得到以下错误，但是，如果要删除字符串，一切都会正常运行。如果其他人遇到这样的错误，请让我知道如何解决！
这就是我的API的样子：
@router.post("/post")
async def post_request(
db: Session = Depends(database.get_db(string_here="string")
):
return ...
和一个数据库文件
def get_db(string_here: str):
.... returns a db
错误：
raise T
浏览 4提问于2022-10-04得票数 0
1回答
在我的烧瓶应用程序中，我有以下代码：
def get_db():
conn =sqlite3.connect(db_path)
cur =conn.cursor()
cur.execute("""CREATE TABLE IF NOT EXISTS Files (name TEXT,
year TEXT, subject TEXT, kind TEXT, correction INT,
description TEXT, date TEXT)""")
return cur
def query_
浏览 5提问于2013-09-21得票数 0
回答已采纳
我使用SQlalchemy连接到MySQL项目中的MySQL服务器，它由实际的API和使用arq编写的后端工作人员组成。两者共享相同的代码基、模型，因此也共享相同的数据库代码。
实际上，我创建的数据库连接如下：
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
import config as c
SQLALCHEMY_DATABASE_URL = "%s://%s:%s@%s:%d/%s" % (c.settings.db_type, c.settings.db_user
浏览 5提问于2022-02-02得票数 0
回答已采纳
2回答
我正在尝试通过Flask将SQLite查询的结果发送到Jinja2模板。为此，我对查询执行.fetchall()，然后将结果作为参数传递给render_template()。在html模板上，我遍历传递的每一行，以便显示每一行的“title”属性。预期的行为是将通过render_template()传递的数据填充到最终的html中。我非常确定我已经完全按照官方Flask文档中的教程项目所显示的方式完成了这项工作。 这是从数据库获取行的代码。 @bp.route('/', methods=('GET',))
def archive():
""&#
浏览 38提问于2020-07-24得票数 0
我正在尝试使用ssl连接mongodb，如下所示，连接mongo会提示Enter PEM pass phrase:，如何避免这个提示？ from pymongo import MongoClient
database_name = "iso_change_life_cycle_qa"
collection_name= "mymongocollection"
def get_db():
try:
# Following are the paths for certificates created by PIE environment. As lo
浏览 30提问于2019-06-18得票数 1
回答已采纳
1回答
我刚接触到Flask，它尝试实现MySQL连接池，而不使用使用SQLAlchemy。我做了一些研究，并了解了如何使用mysql-连接器-python定义连接池(8.0.27)。以下来自官方网站的MySQL提供了一个示例。
dbconfig = {
"database": "test",
"user": "joe"
}
cnxpool = mysql.connector.pooling.MySQLConnectionPool(pool_name = "mypool",
浏览 21提问于2022-01-05得票数 0
回答已采纳
1回答
下面的代码应该是一个简单的用户注册。
database.db
import sqlite3
conn = sqlite3.connect('database.db')
conn.execute('CREATE TABLE accounts (username TEXT, password TEXT)')
conn.close()
views.py
import sqlite3
DATABASE = '../FlaskWebProject/database.db'
with app.app_context():
浏览 35提问于2018-08-17得票数 0
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
实时音视频
对象存储
即时通信 IM活动推荐
腾讯云开发者
