Viktor Andriichuk's questions

大家好。

有这样一个方法：

async with uow:
    logger.info(f"Deleting level {level_id} for project {project_id} started")

    # Check if current_user is the project owner
    logger.info(f"Check if current_user is the project owner started")
    project = await projects_service.get_project_details_by_project_id(
        uow,
        email,
        project_id,
    )
    logger.info(f"Check if current_user is the project owner finished")

    logger.info(f"uow.users_repository.find_one started")
    user = await uow.users_repository.find_one(email=email)
    if not user:
        raise HTTPException(status_code=404, detail="Access denied")
    logger.info(f"uow.users_repository.find_one finished")

    logger.info(f"project.owner_id != user.id check started")
    if project.owner_id != user.id:
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN,
            detail="Only project owner can delete levels from the project",
        )
    logger.info(f"project.owner_id != user.id check finished")

    # Check if the level exists in the project
    logger.info(f"Check if the level exists in the project started")
    level = await uow.projectlevel_repository.find_one(id=level_id, project_id=project_id)
    if not level:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail="Level not found in the project",
        )
    logger.info(f"Check if the level exists in the project finished")

    # First set to NULL for all users with this level
    logger.info(f"Setting NULL for all users with level {level_id} in project {project_id}")
    updated_count = await uow.projectusers_repository.update_many_set_level_null(project_id, level_id)
    logger.info(f"Updated {updated_count} users")

    # Check the level before deletion
    logger.info(f"Checking level {level_id} before deletion")
    level_before = await uow.projectlevel_repository.find_one(id=level_id)
    logger.info(f"Level before deletion: {level_before}")

    # Delete the level
    logger.info(f"Deleting the level {level_id}")
    deleted_level = await uow.projectlevel_repository.delete_one(level_id)
    logger.info(f"Deletion result: {deleted_level}")

    # Immediately check if the level still exists
    logger.info(f"Checking if level {level_id} still exists immediately after deletion")
    check_level_immediate = await uow.projectlevel_repository.find_one(id=level_id)
    logger.info(f"Level immediately after deletion: {check_level_immediate}")

    # Flush the session to ensure changes are visible
    logger.info("Flushing the session")
    await uow.flush()

    # Clear session cache
    await uow.expunge_all()

    # Check again after flush
    logger.info(f"Checking if level {level_id} still exists after flush")
    check_level_after_flush = await uow.projectlevel_repository.find_one(id=level_id)
    logger.info(f"Level after flush: {check_level_after_flush}")

    if check_level_after_flush:
        logger.error(f"Level {level_id} still exists in the database after deletion attempt and flush")
        raise Exception(f"Failed to delete level {level_id}")
    else:
        logger.info(f"Level {level_id} successfully deleted and not found in the database after flush")

    # We get all project users to recalculate levels
    logger.info(f"Getting all project users to recalculate levels")
    project_users, _ = await uow.projectusers_repository.find_all(project_id=project_id, get_all=True)
    logger.info(f"Found {len(project_users)} users for level recalculation")

    # Recalculate levels for all users in the project
    logger.info(f"Recalculating levels for all users in the project")
    for project_user in project_users:
        new_level_id = await calculate_new_level(uow, project_user)
        if new_level_id != project_user.level_id:
            await uow.projectusers_repository.update_one(
                project_user.id,
                {"level_id": new_level_id},
            )
    logger.info(f"Levels recalculated for all users")

    logger.info("Committing changes")
    await uow.commit()
    logger.info("Changes committed successfully")

    # Clear session cache again
    await uow.expunge_all()

    # Final check after commit
    logger.info(f"Final check: Checking if level {level_id} still exists after commit")
    final_check_level = await uow.projectlevel_repository.find_one(id=level_id)
    logger.info(f"Level after commit: {final_check_level}")

    if final_check_level:
        logger.error(f"Level {level_id} still exists in the database after commit")
        raise Exception(f"Failed to delete level {level_id} even after commit")
    else:
        logger.info(f"Level {level_id} successfully deleted and not found in the database after commit")

    return {"message": "Project level successfully deleted and user levels recalculated"}

问题是由于某种原因，该关卡不想被删除：

这是日志：

backend-1  | INFO | July 9, 2024 > 13:52:18 | Level after flush: None
backend-1  | INFO | July 9, 2024 > 13:52:18 | Level 12 successfully deleted and not found in the database after flush

backend-1  | INFO | July 9, 2024 > 13:52:18 | Getting all project users to recalculate levels

backend-1  | INFO | July 9, 2024 > 13:52:18 | Found 2 users for level recalculation

backend-1  | INFO | July 9, 2024 > 13:52:18 | Recalculating levels for all users in the project
backend-1  | INFO | July 9, 2024 > 13:52:18 | Starting level calculation for ProjectUser id=18

backend-1  | INFO | July 9, 2024 > 13:52:18 | Total score for ProjectUser id=18 is 0.00



backend-1  | 2024-07-09 13:52:18,086 INFO sqlalchemy.engine.Engine COMMIT
backend-1  | INFO | July 9, 2024 > 13:52:18 | Found 2 levels for project id=18 and role id=None

backend-1  | INFO | July 9, 2024 > 13:52:18 | ProjectUser id=18 assigned to minimum level_id=11 with level score=50.0

backend-1  | INFO | July 9, 2024 > 13:52:18 | Starting level calculation for ProjectUser id=26


backend-1  | INFO | July 9, 2024 > 13:52:18 | Total score for ProjectUser id=26 is 0.00



backend-1  | INFO | July 9, 2024 > 13:52:18 | Found 2 levels for project id=18 and role id=26

backend-1  | INFO | July 9, 2024 > 13:52:18 | ProjectUser id=26 assigned to minimum level_id=11 with level score=50.0
backend-1  | INFO | July 9, 2024 > 13:52:18 | Levels recalculated for all users

backend-1  | INFO | July 9, 2024 > 13:52:18 | Committing changes

backend-1  | INFO | July 9, 2024 > 13:52:18 | Changes committed successfully

backend-1  | INFO | July 9, 2024 > 13:52:18 | Final check: Checking if level 12 still exists after commit



backend-1  | INFO | July 9, 2024 > 13:52:18 | Level after commit: id=12 created_at=datetime.datetime(2024, 7, 8, 14, 59, 16, 505576) updated_at=datetime.datetime(2024, 7, 8, 14, 59, 16, 505578) project_id=18 role_id=26 name='L6 (Middle+)' score=60.0 salary=3000.0
backend-1  | ERROR | July 9, 2024 > 13:52:18 | Level 12 still exists in the database after commit
backend-1  | 2024-07-09 13:52:18,100 INFO sqlalchemy.engine.Engine ROLLBACK
backend-1  | ERROR | July 9, 2024 > 13:52:18 | Unexpected error: Failed to delete level 12 even after commit
backend-1  | INFO:     192.168.65.1:33388 - "DELETE /api/v1/account/projectlevels/18/12 HTTP/1.1" 500 Internal Server Error

也就是说，在调用删除的代码中间，它似乎在会话中被标记为删除，但是当发生提交时，什么也没有发生。

这是实际执行删除的方法：

async def delete_one(self, entity_id: int) -> bool:
    stmt = delete(self.model).where(self.model.id == entity_id)
    result = await self.session.execute(stmt)
    deleted_count = cast(int, result.rowcount)
    logger.info(f"Deleted {deleted_count} records for level_id {entity_id}")
    return deleted_count > 0

这是工作单元

from abc import ABC, abstractmethod
from types import TracebackType

from backend.db.db import async_session_maker
from backend.repositories.project_user_skill_evaluation import SkillEvaluationRepository
from backend.repositories.project_user_skills_history import (
    ProjectUserSkillsHistoryRepository,
)
from backend.repositories.project_users import ProjectUsersRepository
from backend.repositories.projectlevels import ProjectLevelRepository
from backend.repositories.projectroles import ProjectRoleRepository
from backend.repositories.projects import ProjectsRepository
from backend.repositories.projectskills import ProjectSkillsRepository
from backend.repositories.projectskillsgroup import ProjectSkillsGroupRepository
from backend.repositories.tasks import TasksRepository
from backend.repositories.users import UsersRepository
from backend.utils.span_decorator import trace_span


class IUnitOfWork(ABC):
    users_repository: UsersRepository
    tasks_repository: TasksRepository
    projects_repository: ProjectsRepository
    projectskillsgroup_repository: ProjectSkillsGroupRepository
    projectskills_repository: ProjectSkillsRepository
    projectrole_repository: ProjectRoleRepository
    projectlevel_repository: ProjectLevelRepository
    projectusers_repository: ProjectUsersRepository
    skillevaluation_repository: SkillEvaluationRepository
    projectuserskillhistory_repository: ProjectUserSkillsHistoryRepository

    @abstractmethod
    def __init__(self) -> None:
        pass

    @abstractmethod
    async def __aenter__(self) -> "IUnitOfWork":
        pass

    @abstractmethod
    async def __aexit__(
        self,
        exc_type: type[BaseException] | None,
        exc_val: BaseException | None,
        exc_tb: TracebackType | None,
    ) -> None:
        pass

    @abstractmethod
    async def commit(self) -> None:
        pass

    @abstractmethod
    async def flush(self) -> None:
        pass

    @abstractmethod
    async def rollback(self) -> None:
        pass

    @abstractmethod
    async def expunge_all(self) -> None:
        pass


class UnitOfWork(IUnitOfWork):
    def __init__(self) -> None:
        self.session_factory = async_session_maker

    @trace_span("UnitOfWork: Start")
    async def __aenter__(self) -> "UnitOfWork":
        self.session = self.session_factory()
        self.users_repository = UsersRepository(self.session)
        self.tasks_repository = TasksRepository(self.session)
        self.projects_repository = ProjectsRepository(self.session)
        self.projectskillsgroup_repository = ProjectSkillsGroupRepository(self.session)
        self.projectskills_repository = ProjectSkillsRepository(self.session)
        self.projectrole_repository = ProjectRoleRepository(self.session)
        self.projectlevel_repository = ProjectLevelRepository(self.session)
        self.projectusers_repository = ProjectUsersRepository(self.session)
        self.skillevaluation_repository = SkillEvaluationRepository(self.session)
        self.projectuserskillhistory_repository = ProjectUserSkillsHistoryRepository(
            self.session,
        )
        return self

    @trace_span("UnitOfWork: Exit")
    async def __aexit__(
        self,
        exc_type: type[BaseException] | None,
        exc_val: BaseException | None,
        exc_tb: TracebackType | None,
    ) -> None:
        if exc_type:
            await self.rollback()
        else:
            await self.commit()
        await self.session.close()

    @trace_span("UnitOfWork: Commit")
    async def commit(self) -> None:
        await self.session.commit()

    @trace_span("UnitOfWork: Flush")
    async def flush(self) -> None:
        await self.session.flush()

    @trace_span("UnitOfWork: Rollback")
    async def rollback(self) -> None:
        await self.session.rollback()

    @trace_span("UnitOfWork: Expunge All")
    async def expunge_all(self) -> None:
        self.session.expunge_all()

以防万一，另一个级别模型：

from typing import Any

from sqlalchemy import Float, ForeignKey, Integer, String
from sqlalchemy.orm import Mapped, mapped_column, relationship

from backend.db.db import Base
from backend.models.common import IDMixin, TimestampMixin
from backend.schemas.app.project_levels import ProjectLevelsReadSchema


class ProjectLevels(Base, IDMixin, TimestampMixin):
    __tablename__ = "project_levels"
    project_id: Mapped[int] = mapped_column(
        Integer,
        ForeignKey("projects.id"),
        nullable=False,
    )
    role_id: Mapped[int] = mapped_column(
        Integer,
        ForeignKey("project_roles.id"),
        nullable=False,
    )
    name: Mapped[str] = mapped_column(String, nullable=False)
    score: Mapped[float | None] = mapped_column(Float, nullable=True)
    salary: Mapped[float | None] = mapped_column(Float, nullable=True)

    project = relationship("Project", back_populates="levels")
    role = relationship("ProjectRole", back_populates="levels")
    project_users = relationship("ProjectUser", back_populates="level")

    def to_read_model(
        self,
    ) -> ProjectLevelsReadSchema:
        return ProjectLevelsReadSchema(
            id=self.id,
            project_id=self.project_id,
            role_id=self.role_id,
            name=self.name,
            score=self.score,
            salary=self.salary,
            created_at=self.created_at,
            updated_at=self.updated_at,
        )

    def to_dict(self) -> dict[str, Any]:
        return {
            "id": self.id,
            "project_id": self.project_id,
            "role_id": self.role_id,
            "name": self.name,
            "score": self.score,
            "salary": self.salary,
            "created_at": self.created_at,
            "updated_at": self.updated_at,
        }

使用 great_expectations 时出现问题。

版本 great_expectations==0.13.43 已设置。我尝试使用命令great_expectations checkpoint new，我得到的结果是

命令检查点不存在。请使用以下之一：['checkpoint'、'datasource'、'docs'、'init'、'project'、'store'、'suite'、'validation_operator']

可能是什么问题呢？

基本上，这就是整个问题。我在“Jul2019”（列格式 - 字符串）形式的列中的雪花表中有数据。我需要将此字符串转换为日期。

大家好。

并告诉我如何转换在 Snowflake 下工作但不想在 PostreSQL 下工作的 SQL。对于雪花来说是这样的：

SELECT
    global_user_id
    , site_lookup_prefix
    , COUNT(global_user_id, site_lookup_prefix) AS count_views
FROM cte_page_view

大家好。

有一个请求

SELECT
        global_user_id
        , os_family_1
    FROM (
        SELECT
            MD5(CONCAT(CASE WHEN bums.global_user_id::VARCHAR IS NULL THEN 'NULL'::VARCHAR ELSE bums.global_user_id::VARCHAR END)) AS global_user_id
            , dos.os_family_1 AS os_family_1
            , MAX(fpv.page_view_date) AS max_page_view_date
            , ROW_NUMBER() OVER (PARTITION BY bums.global_user_id ORDER BY fpv.page_view_date DESC) AS rn
        FROM "database"."data_warehouse__prod"."fact_page_views" AS fpv
        LEFT JOIN "database"."data_warehouse__prod"."dim_browser" AS dib ON dib.browser_key = fpv.browser_key
        LEFT JOIN "database"."data_warehouse__prod"."dim_os" AS dos ON dos.os_key = fpv.os_key
        LEFT JOIN "database"."data_warehouse__prod"."dim_page" AS dip ON dip.page_key = fpv.page_key
        LEFT JOIN "database"."data_warehouse__prod"."dim_snowplow_user" AS dsu ON dsu.snowplow_user_key = fpv.snowplow_user_key
        LEFT JOIN "database"."data_warehouse__prod"."outrigger_dim_site" AS ods ON ods.site_name = dip.page_app_id
        LEFT JOIN "database"."data_warehouse__prod"."bridge_user_model_snowplow" AS bums ON dsu.snowplow_user_key = bums.snowplow_user_key
        WHERE
            fpv.page_view_date::DATE >= (('2022-02-02'::DATE + -8 * INTERVAL '1 week'))
            AND fpv.page_view_date::DATE < (('2022-02-02'::DATE + 0 * INTERVAL '1 day'))
            AND (dib.is_a_bot) = 'This Is Not A Bot'
            AND (dib.is_internal_traffic) = 'This Is Not Internal Traffic'
            AND (ods.site_id IS NOT NULL)
        group by 1,2
        ) AS os_family_1
    WHERE rn = 1

但由于某种原因，我得到了一个错误。

column "fpv.page_view_date" must appear in the GROUP BY clause or be used in an aggregate function
LINE 18: ...) OVER (PARTITION BY bums.global_user_id ORDER BY fpv.page_v...

我自己想不通。

大家好。

我有这个 docker-compose：

version: '3.6'

services:
  redis2:
    image: redis:5
    ports:
      - ${PORT}:6379

我用 cli 运行它：docker-compose run -e PORT=6379 -d redis2

我收到一个错误：未设置“PORT”变量。默认为空字符串。

但是什么？这应该根据文档工作......

我正在将代码上传到 Storage Digital Ocean，我需要立即公开此文件。默认情况下它是私有的。

我这样加载：

session = boto3.session.Session()
    client = session.client(
        's3',
        region_name='fra1',
        endpoint_url='https://fra1.digitaloceanspaces.com',
        aws_access_key_id='',
        aws_secret_access_key='',
    )

    # # create new bucket
    # client.create_bucket(Bucket='your-bucket-name')

    # upload file
    with open(video, 'rb') as video:
        client.put_object(
            Bucket='your-bucket-name',
            Key='test.mp4',
            Body=video,
        )

大家好。

我有一个像这样的numpy数组：

array([16, 17, 1, 3, 8, 10, 2, 19, 14, 18, 15, 6, 11, 13, 5, 4, 9, 0, 7, 12])

我需要首先获取每个元素的索引以及值，然后对生成的元素列表进行排序，以便得到如下内容：

[19: 7, 18: 9, 17: 1, 16: 0, 15: 10 ...]

也就是说，在最终列表中，我必须从第一个列表中按递减值进行排序，但同时我需要添加每个元素的基本列表的索引。你需要以最快的方式完成它。我有一个想法如何做到这一点，但不知何故很难。也许 numpy 有更快的方法？

在 RedShift 中，我尝试应用这样的转换（即，为空值放置零）

CASE
    WHEN s.files_duration != '' THEN s.files_duration
    ELSE 0.0 
END :: NUMERIC AS files_duration_2

对于包含点之后的值的值，我希望这些值保持不四舍五入（原样），但由于某种原因它们会被四舍五入。我不明白为什么。

你能建议我做错了什么吗？

谢谢你。

大家好。

你能告诉我为什么在尝试将字符串转换为日期时间时会出现这样的错误吗？

ValueError: time data '2021:6:22' does not match format '%Y:-%m:-%d'

我这样组织翻译：

count_data_for_interval_clicks_dict = [{'count': int(row[0]), 'date': datetime.strptime(str(int(row[1]))+':'+str(int(row[2]))+':'+str(int(row[3])), '%Y:-%m:-%d')} for row in count_data_for_interval_clicks.all()][:10]

当我使用它时：'%Y:%-m:%-d' 它说'-'是格式为'%Y:%-m:%-d'的错误指令。

但似乎文档说我可以分别使用 %-m, %-d if Month 作为十进制数和 Day of the month 作为十进制数，就像我的情况一样。

大家好。

我需要在 RedShift 中解析 JSON。

要解析的字符串位于“inputs”列中，包含以下内容：

[{"desc": "How many people does the video contain?", "name": "Number of People", "type": "dropdown", "values": ["", "Only 1", "2-3", "3+"]}, {"desc": "What is the camera position?", "name": "Movement", "type": "dropdown", "values": ["", "Fixed position", "Moving"]}, {"desc": "From which angle did you shoot the video?", "name": "Shoot Angle", "type": "dropdown", "values": ["", "Frontal recording", "Tight angle: 10-40 degree", "Wide angle: 40-70 degree"]}, {"desc": "From which distance did you shoot the video?", "name": "Distance", "type": "dropdown", "values": ["", "Near/Selfie", "Mid (3-6 ft)", "Far (>6 ft)"]}, {"desc": "What is the video lighting direction?", "name": "Lighting Direction", "type": "dropdown", "values": ["", "Front lit", "Side lit", "Back lit"]}, {"desc": "What is the video background?", "name": "Background", "type": "dropdown", "values": ["", "Outdoors", "In office", "At home", "Plain background"]}, {"desc": "What is the topic in your speech?", "name": "Topic", "type": "dropdown", "values": ["", "Arts and Media", "Business", "Education", "Entertainment", "Food/Eating", "Nutrition", "Healthcare ", "High School Life", "Mental Health", "News", "Technology", "Morals and Ethics", "Phones and Apps", "Sports", "Science"]}]

我需要以一种可以提取每个唯一值并与它们形成单独行的方式解析 JSON。

例如：

id: 1, desc: "How many people does the video contain?" 
id: 2, desc: "What is the camera position?"
etc.

我尝试查询：

SELECT c.*, d.desc, d.name, d.values FROM source.table AS c, c.inputs AS d; 

我收到一个错误：不允许在“输入”列上导航，因为它不是超级类型

我在官方 RedShift 文档中阅读了此查询：https ://docs.aws.amazon.com/redshift/latest/dg/query-super.html

但是根据我目前的数据，它不能正常工作。

当我尝试应用他们文档中描述的内容并在那里添加我的数据中的一行时

CREATE TABLE test_parse_json_super
(
  id smallint,
  details super
);
 
INSERT INTO test_parse_json_super VALUES(1, JSON_PARSE('[{"desc": "How many people does the video contain?", "name": "Number of People", "type": "dropdown", "values": ["", "Only 1", "2-3", "3+"]}, {"desc": "What is the camera position?", "name": "Movement", "type": "dropdown", "values": ["", "Fixed position", "Moving"]}, {"desc": "From which angle did you shoot the video?", "name": "Shoot Angle", "type": "dropdown", "values": ["", "Frontal recording", "Tight angle: 10-40 degree", "Wide angle: 40-70 degree"]}, {"desc": "From which distance did you shoot the video?", "name": "Distance", "type": "dropdown", "values": ["", "Near/Selfie", "Mid (3-6 ft)", "Far (>6 ft)"]}, {"desc": "What is the video lighting direction?", "name": "Lighting Direction", "type": "dropdown", "values": ["", "Front lit", "Side lit", "Back lit"]}, {"desc": "What is the video background?", "name": "Background", "type": "dropdown", "values": ["", "Outdoors", "In office", "At home", "Plain background"]}, {"desc": "What is the topic in your speech?", "name": "Topic", "type": "dropdown", "values": ["", "Arts and Media", "Business", "Education", "Entertainment", "Food/Eating", "Nutrition", "Healthcare ", "High School Life", "Mental Health", "News", "Technology", "Morals and Ethics", "Phones and Apps", "Sports", "Science"]}]'));
                                               

并使用文档中的查询"SELECT c.*, d.desc, d.name, d.values FROM test_parse_json_super AS c, c.details AS d;"

然后它工作并且所有数据正常分解成行（以及JSON本身是正确的）。

我不明白我需要做什么才能使查询对真实表中的真实数据而不是测试表起作用？

也许有人会告诉？

提前致谢！

我正在尝试向 RedShift 添加数据的请求：

-- Insert records
INSERT INTO target.table (
            collect_project_id,
            project_number,
            project_name,
            connect_project_id,
            project_desc,
            project_type,
            project_status,
            project_path,
            language_code,
            country_code,
            timezone,
            date_created,
            date_updated
            )
(SELECT id, number, name, connect_project_id, description, type, status, path, language, country, timezone, created, modified
FROM source.table
LEFT JOIN target.table
ON source.table.id = target.table.collect_project_id 
WHERE target.table.collect_project_id != source.table.projects.id AND (source.table.created > getdate() - interval '30 minutes' OR source.table.modified > getdate() - interval '30 minutes'));

我收到一个错误：

psycopg2.errors.AmbiguousColumn: column reference "connect_project_id" is ambiguous

我有一个这样的查询：

WITH all_values AS (
    SELECT s.projectid AS projectid,
           s.prompttype AS prompttype,
           (s.attributes::jsonb)->>'prompt' AS prompt,
           (s.attributes::jsonb)->>'description' AS description,
           (s.attributes::jsonb)->>'topic' AS topic,
           (s.attributes::jsonb)->>'context' AS context,
           (s.attributes::jsonb)->>'use_case' AS use_case,
           (s.attributes::jsonb)->>'subtitle' AS subtitle,
           (s.attributes::jsonb)->>'txValues' AS txValues,
           (s.attributes::jsonb)->>'flashmode' AS flashmode,
           (s.attributes::jsonb)->>'skippable' AS skippable,
           (s.attributes::jsonb)->>'videoMaxDuration' AS videoMaxDuration,
           (s.attributes::jsonb)->>'defaultCameraFacing' AS defaultCameraFacing,
           s.corpuscode AS corpuscode,
           s.scriptid AS scriptid,
           s.promptnum AS promptnum,
           (SELECT ARRAY_AGG(f) 
               FROM (
                   SELECT jsonb_object_keys(attributes::jsonb) f
                   FROM source_table
                ) AS u) AS others_key
    FROM source_table AS s
    WHERE
        s.prompttype != 'input' AND (s.created > now() - interval '30 minutes' OR s.modified > now() - interval '30 minutes')
    GROUP BY s.projectid, s.prompttype, prompt, description, topic, context, use_case, subtitle, txValues, flashmode, skippable, videoMaxDuration, defaultCameraFacing, corpuscode, scriptid, promptnum
   )
SELECT * FROM all_values;

在 others_key 列中，我只想显示给定 JSON 中的一般键列表中的一些键（即，不包括列表中的键）。我不知道该怎么做。现在，所有键都简单地显示出来。

大家好。

我在如何将 JSON 中的密钥收集到一行（进行连接）时遇到问题。

请求是这样的：

WITH all_values AS (
    SELECT s.projectid AS projectid,
           s.prompttype AS prompttype,
           (s.attributes::jsonb)->>'prompt' AS prompt,
           (s.attributes::jsonb)->>'description' AS description,
           (s.attributes::jsonb)->>'topic' AS topic,
           (s.attributes::jsonb)->>'context' AS context,
           (s.attributes::jsonb)->>'use_case' AS use_case,
           (s.attributes::jsonb)->>'subtitle' AS subtitle,
           (s.attributes::jsonb)->>'txValues' AS txValues,
           (s.attributes::jsonb)->>'flashmode' AS flashmode,
           (s.attributes::jsonb)->>'skippable' AS skippable,
           (s.attributes::jsonb)->>'videoMaxDuration' AS videoMaxDuration,
           (s.attributes::jsonb)->>'defaultCameraFacing' AS defaultCameraFacing,
           s.corpuscode AS corpuscode,
           s.scriptid AS scriptid,
           s.promptnum AS promptnum,
           SELECT ARRAY_AGG(f) 
               FROM (
                   SELECT jsonb_object_keys(attributes::jsonb) f
                   FROM source_table
           ) AS u
    FROM source_table AS s
    WHERE
        s.prompttype != 'input' AND (s.created > now() - interval '30 minutes' OR s.modified > now() - interval '30 minutes')
    GROUP BY s.projectid, s.prompttype, prompt, description, topic, context, use_case, subtitle, txValues, flashmode, skippable, videoMaxDuration, defaultCameraFacing, corpuscode, scriptid, promptnum
   )
SELECT * FROM all_values;

错误：错误：“SELECT”第 18 行或附近的语法错误：SELECT ARRAY_AGG(f)

我这样检查：

SELECT ARRAY_AGG(f) 
FROM (
  SELECT jsonb_object_keys(attributes::jsonb) f
  FROM source_table
) AS u;

并获得输出。

在这里测试：https ://dbfiddle.uk/?rdbms=postgres_9.6&fiddle=dcf7722ab857c13c2d3c8edf172b434e

PS 理想情况下，在这个输出中，我不需要过滤所有的键，而只需要过滤其中的一部分，但这是下一个问题......

大家好。

我需要在 PostregSQL 中解析 JSON。

这是数据表：

CREATE TABLE source_table (
    id              INT,
    prompttype      VARCHAR(20),
    corpuscode      VARCHAR(2000),
    text            VARCHAR(2000),
    attributes      VARCHAR(2000),
    inputs          VARCHAR(2000),
    comment         VARCHAR(2000),
    created         TIMESTAMP WITHOUT TIME ZONE DEFAULT NOW(),
    projectid       BIGINT,
    promptnum       INT,
    scriptid        INT,
    modified        TIMESTAMP WITHOUT TIME ZONE


);

INSERT INTO source_table VALUES (209687,'recording','24-Z-02-669-0001-012','<html><p>Please read the following prompt: </p><p><span style="color: #ff0000;"><strong><span style="font-size: 14pt;">के म जुलाई पन्ध्रमा बस्ने होटल जाँच गर्नुभएको छ ?</span></strong></span></p></html>','{"prompt": "के म जुलाई पन्ध्रमा बस्ने होटल जाँच गर्नुभएको छ ?", "skippable": false, "sampleRate": 16000, "audioMaxDuration": 60, "audioMinDuration": 4, "locationRequired": true, "maxLeadingSilence": 5, "minLeadingSilence": 0.6, "maxTrailingSilence": 5, "minTrailingSilence": 0.6}',null,'','2021-09-01 05:42:15.807260',224,73,10217,'2021-09-03 13:15:15.807260');

SELECT * FROM source_table WHERE prompttype ='input' ORDER BY created DESC LIMIT 10;

CREATE TABLE target_table (
    location_required       BOOLEAN,
    access_medialibrary     BOOLEAN,
    skippable               BOOLEAN,
    flashmode               BOOLEAN,
    other_attributes        VARCHAR(2000),
    custom_framerate        VARCHAR(2000),
    video_quality           VARCHAR(2000),
    default_camerafacing    VARCHAR(2000),
    prompt_values           VARCHAR(2000),
    subtitle                VARCHAR(2000),
    use_case                VARCHAR(2000),
    topic                   VARCHAR(2000),
    prompt_desc             VARCHAR(2000),
    prompt_name             VARCHAR(2000),
    prompt_source           VARCHAR(2000),
    prompt_type             VARCHAR(2000),
    max_recordings          BIGINT,
    min_recordings          BIGINT,
    script_id               BIGINT,
    with_transcription      BIGINT,
    video_maxduration       BIGINT,
    prompt_attribute_id     BIGSERIAL PRIMARY KEY,
    collect_project_id      BIGINT,
    edw_date_updated        TIMESTAMP WITHOUT TIME ZONE,
    edw_date_created        TIMESTAMP WITHOUT TIME ZONE,
    date_updated            TIMESTAMP WITHOUT TIME ZONE,
    date_created            TIMESTAMP WITHOUT TIME ZONE NOT NULL DEFAULT NOW(),
    CONSTRAINT unique_dim_collect_prompt_attr_configs UNIQUE (collect_project_id, prompt_values)

);

这是一个解析器：

WITH all_values AS (
    SELECT s.projectid AS projectid,
           s.prompttype AS prompttype,
           el.attributes->>'prompt' AS prompt,
           el.attributes->>'description' AS description,
           el.attributes->>'topic' AS topic,
           el.attributes->>'context' AS context,
           el.attributes->>'use_case' AS use_case,
           el.attributes->>'subtitle' AS subtitle,
           el.attributes->>'txValues' AS txValues,
           el.attributes->>'flashmode' AS flashmode,
           el.attributes->>'Isskippable' AS Isskippable,
           el.attributes->>'videoMaxDuration' AS videoMaxDuration,
           el.attributes->>'defaultCameraFacing' AS defaultCameraFacing,
           s.corpuscode AS corpuscode,
           s.scriptid AS scriptid,
           s.promptnum AS promptnum
    FROM source_table AS s
       CROSS JOIN LATERAL jsonb_array_elements(s.attributes::jsonb) AS el(attributes)
    WHERE
        s.prompttype != 'input' AND (s.created > now() - interval '30 minutes' OR s.modified > now() - interval '30 minutes')
    GROUP BY s.projectid, s.prompttype, prompt, description, topic, context, use_case, subtitle, txValues, flashmode, Isskippable, videoMaxDuration, defaultCameraFacing, corpuscode, scriptid, promptnum
   )
SELECT * FROM all_values;

抛出一个错误：

错误：无法从对象中提取元素

我不知道出了什么问题。

在这里测试：https ://dbfiddle.uk/?rdbms=postgres_9.6&fiddle=ff08ce40b358b8041e23cd0ee73866ff

将数据从一种模式传输到另一种模式时，必须满足以下条件：

prompt_input_value_id
Generate a unique sequence number for the combination of each prompt_input_value and collect_project_id

我有以下查询工作正常并在 PostgreSQL 中完成工作：

INSERT INTO target.table AS t (
        collect_project_id,
        prompt_type,
        prompt_input_desc,
        prompt_input_name,
        prompt_input_value,
        script_id,
        corpuscode)
    SELECT s.projectid,
           max(s.prompttype),
           max(el.inputs->>'name') AS name,
           max(el.inputs->>'desc') AS description,
           v.value,
           max(s.scriptid),
           max(s.corpuscode)
    FROM source.table AS s
       CROSS JOIN LATERAL jsonb_array_elements(s.inputs::jsonb) AS el(inputs)
       CROSS JOIN LATERAL jsonb_array_elements(el.inputs->'values') AS v(value)
    WHERE
        s.prompttype = 'input' AND (s.created > now() - interval '30 minutes' OR s.modified > now() - interval '30 minutes')
    GROUP BY s.projectid, v.value
    ON CONFLICT
        (prompt_input_value, collect_project_id)
    DO UPDATE SET
        (prompt_input_desc, prompt_input_name, date_updated) =
        (EXCLUDED.prompt_input_desc,
        EXCLUDED.prompt_input_name,
        NOW())
    WHERE t.prompt_input_desc != EXCLUDED.prompt_input_desc
        OR t.prompt_input_name != EXCLUDED.prompt_input_name;

我需要以接受 RedShift 的方式重新制作它。例如，UPSERT (ON CONFLICT) RedShift 不接受。

因此，我想将基本查询分为两部分 - 添加数据和更新。

我通过 WITH 语句添加：

    WITH all_values AS (
    SELECT s.projectid AS projectid,
           max(s.prompttype) AS prompttype,
           max(el.inputs->>'name') AS name,
           max(el.inputs->>'desc') AS description,
           v.value AS value,
           max(s.scriptid) AS scriptid,
           max(s.corpuscode) AS corpuscode
    FROM source.table AS s
       CROSS JOIN LATERAL jsonb_array_elements(s.inputs::jsonb) AS el(inputs)
       CROSS JOIN LATERAL jsonb_array_elements(el.inputs->'values') AS v(value)
    WHERE
        s.prompttype = 'input' AND (s.created > now() - interval '30 minutes' OR s.modified > now() - interval '30 minutes')
    GROUP BY s.projectid, v.value
   ), unique_items AS (
    SELECT all_values.projectid, all_values.prompttype, all_values.name, all_values.description, all_values.value, all_values.scriptid, all_values.corpuscode
    FROM all_values
    WHERE all_values.projectid != (SELECT collect_project_id FROM target.table) AND all_values.value != (SELECT prompt_input_value FROM target.table)
   )
INSERT INTO target.table (
        collect_project_id,
        prompt_type,
        prompt_input_desc,
        prompt_input_name,
        prompt_input_value,
        script_id,
        corpuscode)
        SELECT s.projectid,
               s.prompttype,
               s.name,
               s.description,
               s.value,
               s.scriptid,
               s.corpuscode
        FROM all_values AS s;

我的逻辑是这样的：

• 首先，我获取所有需要添加的条目（我为字典的每个元素从它们中形成单独的行）
• 我只选择那些在我要添加的表中没有找到两列 prompt_input_value 和 collect_project_id 的值组合的行
• 我将结果写入目标表。

但我收到错误错误：运算符不存在：jsonb <> 字符变化提示：没有运算符与给定名称和参数类型匹配。您可能需要添加显式类型转换。职位：1006

你能澄清我哪里出错了吗？

提前致谢！

大家好。

AirFlow 1.10 有一个已知的 puckel/docker-airflow:1.10.9 docker 镜像。

version: '2.1'
services:
    redis:
        image: 'redis:5.0.5'
        # command: redis-server --requirepass redispass

    postgres:
        image: postgres:9.6
        environment:
            - POSTGRES_USER=airflow
            - POSTGRES_PASSWORD=airflow
            - POSTGRES_DB=airflow
            - PGDATA=/var/lib/postgresql/data/pgdata
        volumes:
            - ./pgdata:/var/lib/postgresql/data/pgdata
        ports:
            - "5432:5432"

    webserver:
        image: puckel/docker-airflow:1.10.9
        restart: always
        depends_on:
            - postgres
            - redis
        env_file:
            - .env
        environment:
            - LOAD_EX=n
            - FERNET_KEY=46BKJoQYlPPOexq0OhDZnIlNepKFf87WFwLbfzqDDho=
            - EXECUTOR=Celery
            # - POSTGRES_USER=airflow
            # - POSTGRES_PASSWORD=airflow
            # - POSTGRES_DB=airflow
            # - REDIS_PASSWORD=redispass
        volumes:
            - ./airflow/dags:/usr/local/airflow/dags
            # Uncomment to include custom plugins
            - ./airflow/plugins:/usr/local/airflow/plugins
            - ./logs:/usr/local/airflow/logs
            - ./requirements.txt:/requirements.txt
        ports:
            - "8080:8080"
        command: webserver
        healthcheck:
            test: ["CMD-SHELL", "[ -f /usr/local/airflow/airflow-webserver.pid ]"]
            interval: 30s
            timeout: 30s
            retries: 3

    flower:
        image: puckel/docker-airflow:1.10.9
        restart: always
        depends_on:
            - redis
        environment:
            - EXECUTOR=Celery
            # - REDIS_PASSWORD=redispass
        ports:
            - "5555:5555"
        command: flower

    pgadmin-compose:
        image: dpage/pgadmin4
        environment:
            PGADMIN_DEFAULT_EMAIL: "test@gmail.com"
            PGADMIN_DEFAULT_PASSWORD: "test123!"
        ports:
            - 16543:80
        depends_on:
            - postgres

    scheduler:
        image: puckel/docker-airflow:1.10.9
        restart: always
        depends_on:
            - webserver
        volumes:
            - ./airflow/dags:/usr/local/airflow/dags
            # Uncomment to include custom plugins
            - ./airflow/plugins:/usr/local/airflow/plugins
            - ./logs:/usr/local/airflow/logs
            - ./requirements.txt:/requirements.txt
        environment:
            - LOAD_EX=n
            - FERNET_KEY=46BKJoQYlPPOexq0OhDZnIlNepKFf87WFwLbfzqDDho=
            - EXECUTOR=Celery
            # - POSTGRES_USER=airflow
            # - POSTGRES_PASSWORD=airflow
            # - POSTGRES_DB=airflow
            # - REDIS_PASSWORD=redispass
        command: scheduler

    worker:
        image: puckel/docker-airflow:1.10.9
        restart: always
        depends_on:
            - scheduler
        volumes:
            - ./airflow/dags:/usr/local/airflow/dags
            # Uncomment to include custom plugins
            - ./airflow/plugins:/usr/local/airflow/plugins
            - ./logs:/usr/local/airflow/logs
            - ./requirements.txt:/requirements.txt
        environment:
            - FERNET_KEY=46BKJoQYlPPOexq0OhDZnIlNepKFf87WFwLbfzqDDho=
            - EXECUTOR=Celery
            # - POSTGRES_USER=airflow
            # - POSTGRES_PASSWORD=airflow
            # - POSTGRES_DB=airflow
            # - REDIS_PASSWORD=redispass
        command: worker

我不明白我需要添加什么参数才能使用我的登录名和密码登录管理面板。有人解决了这个？

如果发现重复数据，如何在尝试更新行中的数据时正确替换 RedShift 中的 ON CONFLICT 参数？因为据我了解，RedShift 不接受 ON CONFLICT。无论如何，我在这个参数上得到一个错误。

我正在尝试在 RedShift 中提出请求：

INSERT INTO target.table AS t (
            project_id,
            project_number,
            project_name,
            connect_project_id,
            project_desc,
            project_type,
            project_status,
            project_path,
            language_code,
            country_code,
            timezone,
            project_created_at,
            project_modified_at
            )
        SELECT
            s.id,
            s.number,
            s.name,
            s.connect_project_id,
            s.description,
            s.type,
            s.status,
            s.path,
            s.language,
            s.country,
            s.timezone,
            s.created,
            s.modified
        FROM source.table AS s
        ON CONFLICT (project_id)
        DO UPDATE SET
            (project_number, project_name, connect_project_id, project_desc,
            project_type, project_status, project_path, language_code,
            country_code, timezone, project_created_at, project_modified_at,
            date_updated) = (EXCLUDED.project_number,
            EXCLUDED.project_name, EXCLUDED.connect_project_id,
            EXCLUDED.project_desc, EXCLUDED.project_type, EXCLUDED.project_status,
            EXCLUDED.project_path, EXCLUDED.language_code, EXCLUDED.country_code,
            EXCLUDED.timezone, EXCLUDED.project_created_at,
            EXCLUDED.project_modified_at, NOW())
        WHERE
            t.project_number != EXCLUDED.project_number
            OR t.project_name != EXCLUDED.project_name
            OR t.connect_project_id != EXCLUDED.connect_project_id
            OR t.project_desc != EXCLUDED.project_desc
            OR t.project_type != EXCLUDED.project_type
            OR t.project_status != EXCLUDED.project_status
            OR t.project_path != EXCLUDED.project_path
            OR t.language_code != EXCLUDED.language_code
            OR t.country_code != EXCLUDED.country_code
            OR t.timezone != EXCLUDED.timezone
            OR t.project_created_at != EXCLUDED.project_created_at
            OR t.project_modified_at != EXCLUDED.project_modified_at;

但我收到一个错误：错误：“AS”位置或附近的语法错误：42

我不知道出了什么问题。也许有人知道？

我有两张桌子：

CREATE TABLE source.staticprompts (
    id              INT,
    projectid       BIGINT,
    scriptid        INT,
    promptnum       INT,
    prompttype      VARCHAR(20),
    inputs          VARCHAR(2000),
    attributes      VARCHAR(2000),
    text            VARCHAR(2000),
    corpuscode      VARCHAR(2000),
    comment         VARCHAR(2000),
    created         TIMESTAMP,
    modified        TIMESTAMP


);


CREATE TABLE target.dim_collect_user_inp_configs (
    collect_project_id      BIGINT NOT NULL,
    prompt_type             VARCHAR(20),
    prompt_input_desc       VARCHAR(3000),
    prompt_input_name       VARCHAR(1000),
    no_of_prompt_count      BIGINT,
    prompt_input_value      VARCHAR(100),
    prompt_input_value_id   BIGSERIAL PRIMARY KEY,
    script_id               BIGINT,
    corpuscode              VARCHAR(20),
    min_recordings          VARCHAR(2000),
    max_recordings          VARCHAR(2000),
    recordings_count        VARCHAR(2000),
    lease_duration          VARCHAR(2000),
    date_created            TIMESTAMP WITHOUT TIME ZONE NOT NULL DEFAULT NOW(),
    date_updated            TIMESTAMP WITHOUT TIME ZONE,
    CONSTRAINT must_be_unique UNIQUE (prompt_input_value, collect_project_id)

);

我需要将数据从 source.staticprompts 复制到 target.dim_collect_user_inp_configs，如果发现 prompt_input_value 已经有一个值，那么我需要查看 prompt_input_name 和 prompt_input_desc 是否已更改，如果是，请更新它（连同更新日期）。

我正在尝试进行这样的查询：

INSERT INTO target.dim_collect_user_inp_configs AS t (
            collect_project_id,
            prompt_type,
            prompt_input_desc,
            prompt_input_name,
            prompt_input_value,
            script_id,
            corpuscode)
        SELECT
            s.projectid,
            s.prompttype,
            el.inputs->>'name' AS name,
            el.inputs->>'desc' AS description,
            jsonb_array_elements(el.inputs->'values') AS value,
            s.scriptid,
            s.corpuscode
        FROM  source.staticprompts AS s,
            jsonb_array_elements(s.inputs::jsonb) el(inputs)
        ON CONFLICT
            (prompt_input_value)
        DO UPDATE SET
            (prompt_input_desc, prompt_input_name, date_updated) =
            (EXCLUDED.prompt_input_desc,
            EXCLUDED.prompt_input_name,
            NOW())
        WHERE t.prompt_input_desc != EXCLUDED.prompt_input_desc
            OR t.prompt_input_name != EXCLUDED.prompt_input_name
RETURNING *;

但它给出了一个错误：没有与 ON CONFLICT 规范匹配的唯一或排除约束

我无法弄清楚我做错了什么。

在此过程中，问题... RETURNING * 是否有可能仅显示更改影响的总行数？不允许仅返回 count(*)，因为您不能在此处使用聚合函数。