标签: sqlalchemy

在我的 sqlobject 实现中，我有以下行：

members = Member.select("CONCAT_WS(' ', member.first_name, member.last_name, member.personal_code_number, member.mail) like " + Member.sqlrepr('%' + query + '%'))

我想将其转换为使用 sqlalchemy 但尚未找到方法。

我使用以下代码使用 sqlalchemy 创建 postgresql 数据库：

engine=create_engine('postgresql+psycopg2://postgres@localhost/testData')
Base.metadata.create_all(engine)

但即使我在 psql 中手动创建数据库，它也会给我以下错误：

  File "/home/ubuntu/venve/local/lib/python2.7/site-packages/sqlalchemy/engine/default.py", line 376, in connect
return self.dbapi.connect(*cargs, **cparams)
  File "/home/ubuntu/venve/local/lib/python2.7/site-packages/psycopg2/__init__.py", line 164, in connect
  conn = _connect(dsn, connection_factory=connection_factory, async=async)
sqlalchemy.exc.OperationalError: (OperationalError) FATAL:  database "testData" does not exist

为什么会发生这种情况？

我正在使用 SqlAlchemy-1.0.9 和 python 2.7。

因此，当我尝试将对象添加到会话时，没有任何反应，因此数据不会插入数据库中。顺便说一下，我正在使用 mysql 和 oursql。

下面的一些代码：

首先，我创建一个函数来创建“创建引擎”等...并且每次我想创建会话时都会使用它。

base = declarative_base()

engine = create_engine('mysql+oursql://root:toor@127.0.0.1/pandora', echo=True)

def sessionLoader():
    base.metadata.bind = engine
    DBSession = sessionmaker(bind=engine)
    session = DBSession()
    return session

之后，在我的类中，我调用 sessionLoader() 方法：

我创建对象等，然后使用 add() 方法：

newEntity = Entity(name = 'John')

sessionLoader().add(newEntity)

sessionLoader().commit()

之后控制台上不再显示任何错误或异常。但是当我进入数据库并进行一些选择时，那里没有数据。

为了进行测试，我输入了以下代码：

print newEntity in sessionLoader()

它在控制台上显示“False”。

所以'add()'没有做它应该做的工作。

有什么建议或帮助吗？

应用程序昨天在家中在不同的系统上运行良好**当我尝试在不同的系统上运行它时（**新安装的所有东西） -所以我认为这个错误仅适用于最新版本。

目标是将会话中的用户连接到第二个数据库

我收到这个错误： create_engine() missing 1 required positional argument: 'engine_opts'

我目前的代码：

app.config['SQLALCHEMY_DATABASE_URI'] = "postgres:// ****" #main database
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = "True"
app.config['SQLALCHEMY_ENGINE_OPTIONS'] = "None" #i didnt have that line in my system 
before i've just added it in attempt of fixing this issue

db = SQLAlchemy(app,session_options={"autoflush": False})


@app.route('/test')
def test():
    x = 'postgres://*****' #secondary database for user in session only.
    engine = db.create_engine(x)

return ''

我尝试过的事情：

检查库页面： flask_SQLAlchemy Library

我发现的一切是这样的：

create_engine ( sa_url, engine_opts ) …

我使用 bwlow 代码通过 sqlalchemy 连接到 MS SQL，现在它已迁移到 azure cloud。我尝试更改值代码，但我认为这不是连接 ActiveDirectoryPassword 的正确方法

    import sqlalchemy
        from sqlalchemy import event, create_engine
# OLD connection string 
        engine = sqlalchemy.create_engine("mssql+pyodbc://" + "username" + ":" + "passkey" + "@" + "server" + "/" + "Database" + "?driver=SQL+Server"

        @event.listens_for(engine, 'before_cursor_execute')
        def receive_before_cursor_execute(conn, cursor, statement, params, context, executemany):
            if executemany:
                cursor.fast_executemany = True
                cursor.commit()

    # New connection string (for Active directory connection - not working)
    engine = sqlalchemy.create_engine("mssql+pyodbc://" + "abc@domain.com" + ":" + "passkey" + "@" + "xxxx-svsql1.database.windows.net" …

我正在编写一个创建一些 SQLAlchemy 对象的 python 脚本，检查其中哪些对象已经添加到数据库中，然后添加任何新对象。我的脚本如下所示：

from sqlalchemy import Column, String, Integer, ForeignKey, create_engine
from sqlalchemy.orm import relationship, Session
from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()

# Define models
class Person(Base):
    __tablename__ = "Person"
    id = Column(Integer, primary_key=True)
    name = Column(String(100), nullable=False)
    pets = relationship("Pet", backref="person")

    def __repr__(self):
        return f"<Person: {self.name}>"


class Pet(Base):
    __tablename__ = "Pet"
    id = Column(Integer, primary_key=True)
    name = Column(String(100), nullable=False)
    person_id = Column(Integer, ForeignKey("Person.id"))

    def __repr__(self):
        return f"<Pet: {self.name}>"

connection_string = "sqlite:///db.sqlite3"
engine = create_engine(connection_string)
session = …

我想知道这两者有什么区别？我经常看到人们使用 .fetchone() ，而我只将 .first() 放在那里。他们是一样的吗？如果不是，一个比另一个快吗？

我试图让一个网页显示数据库中列出的所有即将上映的节目，以及分别存储在场地和艺术家表中的关于每个节目的一堆信息。我查询 Show 表，然后为该结果中的每一行链接 Artist 和 Venue 表。因为错误有事情做与日期时间，我假设的错误是与任一if show.start_time > current_time:或"start_time": show.start_time但误差不拖住它在哪里。我尝试将 current_time 和 show.start_time 都转换为字符串，看看是否有效，但没有运气。我究竟做错了什么？

@app.route('/shows')
def shows():

  current_time = datetime.now()

  data = []

  allshows = Show.query.all()

  for show in allshows:
    artist = Artist.query.get(show.artist_id)
    venue = Venue.query.get(show.venue_id)

    if show.start_time > current_time:
      data.append({
        "venue_id": show.venue_id,
        "venue_name": venue.name,
        "artist_id": show.artist_id,
        "artist_name": artist.name,
        "artist_image_link": artist.image_link,
        "start_time": show.start_time
      })
 
  return render_template('pages/shows.html', shows=data)

如果有帮助，以下是与之相关的模型：

class Venue(db.Model):
    __tablename__ = 'venues'

    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String)
    city = db.Column(db.String(120))
    state = …

我将 Flask-SQLAlchemy 与本地托管的 postgresql 一起使用，我有这段代码，我应该在用户登录后更新数据库中的用户数据

    user_id = current_user.get_id()
    print(f"\n\n user_id: {user_id}\n\n") #1--->
    x = users.query.filter_by(id=user_id).first()
    x.sid = request.sid
    db.session.commit()
    print(f'\n\n user_object.sid:{x.sid} , request.sid:{request.sid}  \n\n') #2 --->

#1 : userid:1

#2 : x.sid:QKfGvdsL-HGUM-FHHYF, request.sid:QKfGvdsL-HGUM-FHHYF

但是当我查看 pgadmin4 或稍后请求 x.sid 的值时，它总是“空”

我四处看了看，每个人似乎都以与我编写代码的方式相同的方式建议。

值得一提 ：

1. sid column 已使用flask.migrate 添加，尚未与表一起创建。
2. 只是为了确保我的代码中没有任何内容覆盖值后缀，pgadmin4 将空值覆盖为唯一的内容并保持不变，其值没有发生任何变化。
3. 我尝试使用db.session.add(x)但它一直说数据信息已经存在

我正在尝试使用 FastAPI 中的 PUT 操作更新数据库中的单个记录。但出于某种原因，我不断收到此错误。除了这一项之外，所有其他操作都可以正常工作。该错误仅针对更新查询引发。

AttributeError: 'Blog' object has no attribute 'items'

这是相关的代码。

def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

class Blog(BaseModel):
    title: str
    body: str

@app.put('/blog/{id}', status_code=status.HTTP_204_NO_CONTENT, response_class=Response)
def update(id: int, request: schemas.Blog, db: Session = Depends(get_db)):
    blog = db.query(models.Blog).filter(models.Blog.id == id)
    if not blog.first():
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND,
                            detail=f'Blog with id {id} not found')
    blog.update(request)
    db.commit()

这是堆栈跟踪：

Traceback (most recent call last):
  File "c:\dev\fast-tuts\env\lib\site-packages\uvicorn\protocols\http\h11_impl.py", line 396, in run_asgi        
    result = await app(self.scope, self.receive, self.send)
  File "c:\dev\fast-tuts\env\lib\site-packages\uvicorn\middleware\proxy_headers.py", line …