我正在尝试使用 Postgres 数据库构建 FastApi 应用程序，并使用 SQLAlchemy 和 Alembic（分别用于处理数据库建模和迁移）。这将是一个 GraphQL API，所以我使用石墨烯来实现。
这是我的文件：-
models.pyfrom sqlalchemy import Column, DateTime, Integer, Enum, String, Boolean, JSON
from sqlalchemy.orm import column_property, validates, relationship
from sqlalchemy.sql.schema import ForeignKey
from sqlalchemy.sql import func
from config import DEFAULT_AVATARS
from database import Base
import enum
# Status Choices for User's Membership Status
class MembershipStatusChoice(enum.Enum):
UI = 'UNINITIALIZED'
PE = 'PENDINIG'
AP = 'APPROVED'
SU = 'SUSPENDED'
# End of Status Choices
# Start of setting up Course status choices
class CourseStatusChoice(enum.Enum):
DR = 'DRAFT'
PU = 'PUBLISHED'
AR = 'ARCHIVED'
# End of status choices
class AnnouncementsSeen(Base):
__tablename__ = 'announcements_seen'
id = Column(Integer, primary_key=True)
user_id = relationship('User', backref='user')
announcement_id = Column(ForeignKey('announcement.id'), index=True)
class CompletedCourses(Base):
__tablename__ = 'completed_courses'
id = Column(Integer, primary_key=True)
participant_id = relationship('User', backref='user')
course_id = Column(ForeignKey('course.id'), index=True)
class CompletedChapters(Base):
__tablename__ = 'completed_chapters'
id = Column(Integer, primary_key=True)
participant_id = relationship('User', backref='user')
chapter_id = Column(ForeignKey('chapter.id'), index=True)
course_id = Column(ForeignKey('course.id'))
status = Column(Enum(CourseStatusChoice))
scored_points = Column(Integer, default=0)
total_points = Column(Integer, default=0)
percentage = Column(Integer, default=0)
class User(Base):
__tablename__ = "user"
id = Column(Integer, primary_key=True, index=True)
firstname = Column(String(50), default="Uninitialized", nullable=False)
lastname = Column(String(50), default="User", nullable=False)
name = column_property(firstname + " " + lastname)
email = Column(String(255), nullable=False, unique=True)
avatar = Column(String(250), nullable=False, default=DEFAULT_AVATARS['USER'])
institution_id = Column(Integer, ForeignKey('institution.id'), nullable=False)
role = Column(String, ForeignKey('user_role.name'), nullable=True)
title = Column(String(150), nullable=True)
bio = Column(String(300), nullable=True)
@validates('email')
def validate_email(self, key, email):
if isinstance(email, str):
return email.lower()
return email
membership_status = Column(Enum(MembershipStatusChoice), nullable=False, default='SU')
chapters = relationship('CompletedChapters', backref='completed_chapters')
courses = relationship('CompletedCourses', backref='completed_courses')
announcements = relationship("AnnouncementsSeen", backref='announcements_seen')
searchField = Column(String(1000))
active = Column(Boolean, default=True)
created_at = Column(
DateTime(timezone=True),
server_default=func.now()
)
updated_at = Column(
DateTime(timezone=True),
server_default=func.now(),
server_onupdate=func.now()
)
class UserRole(Base):
__tablename__ = 'user_role'
name = Column(String(50), primary_key=True, unique=True, index=True)
description = Column(String(500))
priority = Column(Integer) # Lower the number higher the priority
def default_permissions():
return {}
permissions = Column(JSON, default=default_permissions)
searchField = Column(String(600))
active = Column(Boolean, default=True)
created_at = Column(
DateTime(timezone=True),
server_default=func.now()
)
updated_at = Column(
DateTime(timezone=True),
server_default=func.now(),
server_onupdate=func.now()
)
schemas.pyfrom graphene_sqlalchemy import SQLAlchemyObjectType
from pydantic import BaseModel
from models import User
class UserSchema(BaseModel):
title: str
content: str
class UserModel(SQLAlchemyObjectType):
class Meta:
model = User
main.pyimport graphene
from models import User
from schemas import UserModel
from database import db_session
db = db_session.session_factory()
class Query(graphene.ObjectType):
all_users = graphene.List(UserModel)
user_by_id = graphene.Field(UserModel, user_id=graphene.Int(required=True))
def resolve_all_users(self, info):
query = UserModel.get_query(info)
return query.all()
def resolve_user_by_id(self, info, user_id):
return db.query(User).filter(User.id == user_id).first()
app = FastAPI()
app.add_route("/graphql", GraphQLApp(schema=graphene.Schema(query=Query)))
我把这一切都放在泊坞窗里了。因此，我使用这些命令进行迁移并运行应用程序：-
"makemigrations": "docker-compose run app alembic revision --autogenerate -m 'New Migration'",
"migrate": "docker-compose run app alembic upgrade head",
运行应用程序的 docker 命令是
uvicorn main:app --host 0.0.0.0 --port 8000 --reload但是当我进行迁移并迁移并启动应用程序时，这就是我得到的：-
app | INFO: Uvicorn running on http://0.0.0.0:8000 (Press CTRL+C to quit)
app | INFO: Started reloader process [1] using statreload
app | Process SpawnProcess-1:
app | Traceback (most recent call last):
app | File "/usr/local/lib/python3.8/multiprocessing/process.py", line 315, in _bootstrap
app | self.run()
app | File "/usr/local/lib/python3.8/multiprocessing/process.py", line 108, in run
app | self._target(*self._args, **self._kwargs)
app | File "/usr/local/lib/python3.8/site-packages/uvicorn/subprocess.py", line 76, in subprocess_started
app | target(sockets=sockets)
app | File "/usr/local/lib/python3.8/site-packages/uvicorn/server.py", line 50, in run
app | loop.run_until_complete(self.serve(sockets=sockets))
app | File "/usr/local/lib/python3.8/asyncio/base_events.py", line 616, in run_until_complete
app | return future.result()
app | File "/usr/local/lib/python3.8/site-packages/uvicorn/server.py", line 57, in serve
app | config.load()
app | File "/usr/local/lib/python3.8/site-packages/uvicorn/config.py", line 318, in load
app | self.loaded_app = import_from_string(self.app)
app | File "/usr/local/lib/python3.8/site-packages/uvicorn/importer.py", line 22, in import_from_string
app | module = importlib.import_module(module_str)
app | File "/usr/local/lib/python3.8/importlib/__init__.py", line 127, in import_module
app | return _bootstrap._gcd_import(name[level:], package, level)
app | File "<frozen importlib._bootstrap>", line 1014, in _gcd_import
app | File "<frozen importlib._bootstrap>", line 991, in _find_and_load
app | File "<frozen importlib._bootstrap>", line 975, in _find_and_load_unlocked
app | File "<frozen importlib._bootstrap>", line 671, in _load_unlocked
app | File "<frozen importlib._bootstrap_external>", line 843, in exec_module
app | File "<frozen importlib._bootstrap>", line 219, in _call_with_frames_removed
app | File "/app/api/./main.py", line 6, in <module>
app | from schemas import UserModel, UserSchema
app | File "/app/api/./schemas.py", line 12, in <module>
app | class UserModel(SQLAlchemyObjectType):
app | File "/usr/local/lib/python3.8/site-packages/graphene/utils/subclass_with_meta.py", line 52, in __init_subclass__
app | super_class.__init_subclass_with_meta__(**options)
app | File "/usr/local/lib/python3.8/site-packages/graphene_sqlalchemy/types.py", line 210, in __init_subclass_with_meta__
app | assert is_mapped_class(model), (
app | AssertionError: You need to pass a valid SQLAlchemy Model in UserModel.Meta, received "<class 'models.User'>".
我发现这些 GitHub 问题 表明 sqlalchemy 发出的静默错误可能没有显示在堆栈跟踪中。但我无法使用该信息来精确查明我的模型配置出了什么问题。我尝试解决那里的多对多关系，甚至删除它们以查看是否可以解决问题。但它一直显示相同的错误。需要帮助！
嗨，我认为错误即将到来，因为在 main.py 中你还没有调用元数据。
class User(SQLAlchemyObjectType):
"""
User Connection
"""
class Meta:
"""setting up connection for flow_details"""
model = models.User
# use `exclude_fields` to exclude specific fields ie "id"
interfaces = (relay.Node,)
# fields = "__all__"
class Query(graphene.ObjectType):
node = relay.Node.Field()
get_user=SQLAlchemyConnectionField(User.connection)
也从石墨烯导入继电器导入。您需要先设置连接。当您调用解析器时，它没有获取任何 Sqlalchemyobject 来从中获取数据
我能够通过注释类定义中的关系来解决这个问题，如下所示。似乎是类定义和关系的问题，接下来我需要研究一下。
class Department(Base):
__table_args__ = {"schema":"dbo"}
__tablename__ = 'graphql_department_temp'
id = Column(Text, primary_key=True)
name = Column(Text)
class Employee(Base):
__table_args__ = {"schema":"dbo"}
__tablename__ = 'graphql_employee_temp'
id = Column(Text, primary_key=True)
name = Column(Text)
# hired_on = Column(DateTime, default=func.now())
department_id = Column(Text, ForeignKey('department.id'))
*# department = relationship(
# Department,
# backref=backref('employees',
# uselist=True,
# cascade='delete,all'))*